[PATCH v2 00/10] Multifd Migration Compression

[PATCH v2 00/10] Multifd Migration Compression

[Juan Quintela]

[v2]
- rebase on top of previous arguments posted to the list
- introduces zlib compression
- introduces zstd compression

Please help if you know anything about zstd/zlib compression.

This puts compression on top of multifd. Advantages about current
compression:

- We copy all pages in a single packet and then compress the whole
  thing.

- We reuse the compression stream for all the packets sent through the
  same channel.

- We can select nocomp/zlib/zstd levels of compression.

Please, review.

Juan Quintela (10):
  migration: Increase default number of multifd channels to 16
  migration-test: Add migration multifd test
  migration-test: introduce functions to handle string parameters
  migration: Make multifd_save_setup() get an Error parameter
  migration: Make multifd_load_setup() get an Error parameter
  migration: Add multifd-compress parameter
  migration: Make no compression operations into its own structure
  migration: Add zlib compression multifd support
  configure: Enable test and libs for zstd
  migration: Add zstd compression multifd support

 configure                    |  30 ++
 hw/core/qdev-properties.c    |  13 +
 include/hw/qdev-properties.h |   3 +
 migration/migration.c        |  36 +-
 migration/migration.h        |   3 +-
 migration/ram.c              | 750 ++++++++++++++++++++++++++++++++++-
 migration/ram.h              |   4 +-
 migration/rdma.c             |   2 +-
 monitor/hmp-cmds.c           |  13 +
 qapi/migration.json          |  30 +-
 tests/migration-test.c       | 112 ++++++
 11 files changed, 972 insertions(+), 24 deletions(-)

-- 
2.23.0

[PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Juan Quintela]

We can scale much better with 16, so we can scale to higher numbers.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/migration/migration.c b/migration/migration.c
index 354ad072fa..e7f707e033 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -86,7 +86,7 @@
 
 /* The delay time (in ms) between two COLO checkpoints */
 #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY (200 * 100)
-#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 2
+#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 16
 
 /* Background transfer rate for postcopy, 0 means unlimited, note
  * that page requests can still exceed this limit.
-- 
2.23.0

[PATCH v2 02/10] migration-test: Add migration multifd test

[Juan Quintela]

We set multifd-channels.

Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Thomas Huth <thuth@redhat.com>
Tested-by: Wei Yang <richardw.yang@linux.intel.com>
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 tests/migration-test.c | 56 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 56 insertions(+)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index f58430c1cb..1c9f2c4e6a 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -1361,6 +1361,61 @@ static void test_migrate_auto_converge(void)
     test_migrate_end(from, to, true);
 }
 
+static void test_multifd_tcp(void)
+{
+    MigrateStart *args = migrate_start_new();
+    QTestState *from, *to;
+    QDict *rsp;
+    char *uri;
+
+    if (test_migrate_start(&from, &to, "defer", args)) {
+        return;
+    }
+
+    /*
+     * We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter_int(from, "downtime-limit", 1);
+    /* 1GB/s */
+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
+
+    migrate_set_parameter_int(from, "multifd-channels", 16);
+    migrate_set_parameter_int(to, "multifd-channels", 16);
+
+    migrate_set_capability(from, "multifd", "true");
+    migrate_set_capability(to, "multifd", "true");
+
+    /* Start incoming migration from the 1st socket */
+    rsp = wait_command(to, "{ 'execute': 'migrate-incoming',"
+                           "  'arguments': { 'uri': 'tcp:127.0.0.1:0' }}");
+    qobject_unref(rsp);
+
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    uri = migrate_get_socket_address(to, "socket-address");
+
+    migrate(from, uri, "{}");
+
+    wait_for_migration_pass(from);
+
+    /* 300ms it should converge */
+    migrate_set_parameter_int(from, "downtime-limit", 600);
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+    test_migrate_end(from, to, true);
+    free(uri);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1425,6 +1480,7 @@ int main(int argc, char **argv)
                    test_validate_uuid_dst_not_set);
 
     qtest_add_func("/migration/auto_converge", test_migrate_auto_converge);
+    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
 
     ret = g_test_run();
 
-- 
2.23.0

[PATCH v2 03/10] migration-test: introduce functions to handle string parameters

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 tests/migration-test.c | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 1c9f2c4e6a..fc221f172a 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -460,6 +460,43 @@ static void migrate_set_parameter_int(QTestState *who, const char *parameter,
     migrate_check_parameter_int(who, parameter, value);
 }
 
+static char *migrate_get_parameter_str(QTestState *who,
+                                       const char *parameter)
+{
+    QDict *rsp;
+    char *result;
+
+    rsp = wait_command(who, "{ 'execute': 'query-migrate-parameters' }");
+    result = g_strdup(qdict_get_str(rsp, parameter));
+    qobject_unref(rsp);
+    return result;
+}
+
+static void migrate_check_parameter_str(QTestState *who, const char *parameter,
+                                        const char *value)
+{
+    char *result;
+
+    result = migrate_get_parameter_str(who, parameter);
+    g_assert_cmpstr(result, ==, value);
+    g_free(result);
+}
+
+__attribute__((unused))
+static void migrate_set_parameter_str(QTestState *who, const char *parameter,
+                                      const char *value)
+{
+    QDict *rsp;
+
+    rsp = qtest_qmp(who,
+                    "{ 'execute': 'migrate-set-parameters',"
+                    "'arguments': { %s: %s } }",
+                    parameter, value);
+    g_assert(qdict_haskey(rsp, "return"));
+    qobject_unref(rsp);
+    migrate_check_parameter_str(who, parameter, value);
+}
+
 static void migrate_pause(QTestState *who)
 {
     QDict *rsp;
-- 
2.23.0

[PATCH v2 04/10] migration: Make multifd_save_setup() get an Error parameter

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c | 2 +-
 migration/ram.c       | 2 +-
 migration/ram.h       | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index e7f707e033..5a56bd0c91 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -3400,7 +3400,7 @@ void migrate_fd_connect(MigrationState *s, Error *error_in)
         return;
     }
 
-    if (multifd_save_setup() != 0) {
+    if (multifd_save_setup(&error_in) != 0) {
         migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
                           MIGRATION_STATUS_FAILED);
         migrate_fd_cleanup(s);
diff --git a/migration/ram.c b/migration/ram.c
index 38070f1bb2..1f364cc23d 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -1208,7 +1208,7 @@ static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
     }
 }
 
-int multifd_save_setup(void)
+int multifd_save_setup(Error **errp)
 {
     int thread_count;
     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
diff --git a/migration/ram.h b/migration/ram.h
index bd0eee79b6..da22a417ea 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -41,7 +41,7 @@ int xbzrle_cache_resize(int64_t new_size, Error **errp);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_total(void);
 
-int multifd_save_setup(void);
+int multifd_save_setup(Error **errp);
 void multifd_save_cleanup(void);
 int multifd_load_setup(void);
 int multifd_load_cleanup(Error **errp);
-- 
2.23.0

[PATCH v2 05/10] migration: Make multifd_load_setup() get an Error parameter

[Juan Quintela]

We need to change the full chain to pass the Error parameter.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c | 10 +++++-----
 migration/migration.h |  2 +-
 migration/ram.c       |  2 +-
 migration/ram.h       |  2 +-
 migration/rdma.c      |  2 +-
 5 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index 5a56bd0c91..cf6cec5fb6 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -518,11 +518,11 @@ fail:
     exit(EXIT_FAILURE);
 }
 
-static void migration_incoming_setup(QEMUFile *f)
+static void migration_incoming_setup(QEMUFile *f, Error **errp)
 {
     MigrationIncomingState *mis = migration_incoming_get_current();
 
-    if (multifd_load_setup() != 0) {
+    if (multifd_load_setup(errp) != 0) {
         /* We haven't been able to create multifd threads
            nothing better to do */
         exit(EXIT_FAILURE);
@@ -572,13 +572,13 @@ static bool postcopy_try_recover(QEMUFile *f)
     return false;
 }
 
-void migration_fd_process_incoming(QEMUFile *f)
+void migration_fd_process_incoming(QEMUFile *f, Error **errp)
 {
     if (postcopy_try_recover(f)) {
         return;
     }
 
-    migration_incoming_setup(f);
+    migration_incoming_setup(f, errp);
     migration_incoming_process();
 }
 
@@ -596,7 +596,7 @@ void migration_ioc_process_incoming(QIOChannel *ioc, Error **errp)
             return;
         }
 
-        migration_incoming_setup(f);
+        migration_incoming_setup(f, errp);
 
         /*
          * Common migration only needs one channel, so we can start
diff --git a/migration/migration.h b/migration/migration.h
index 79b3dda146..545f283ae7 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -265,7 +265,7 @@ struct MigrationState
 
 void migrate_set_state(int *state, int old_state, int new_state);
 
-void migration_fd_process_incoming(QEMUFile *f);
+void migration_fd_process_incoming(QEMUFile *f, Error **errp);
 void migration_ioc_process_incoming(QIOChannel *ioc, Error **errp);
 void migration_incoming_process(void);
 
diff --git a/migration/ram.c b/migration/ram.c
index 1f364cc23d..fcf50e648a 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -1421,7 +1421,7 @@ static void *multifd_recv_thread(void *opaque)
     return NULL;
 }
 
-int multifd_load_setup(void)
+int multifd_load_setup(Error **errp)
 {
     int thread_count;
     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
diff --git a/migration/ram.h b/migration/ram.h
index da22a417ea..42be471d52 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -43,7 +43,7 @@ uint64_t ram_bytes_total(void);
 
 int multifd_save_setup(Error **errp);
 void multifd_save_cleanup(void);
-int multifd_load_setup(void);
+int multifd_load_setup(Error **errp);
 int multifd_load_cleanup(Error **errp);
 bool multifd_recv_all_channels_created(void);
 bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
diff --git a/migration/rdma.c b/migration/rdma.c
index e241dcb992..2379b8345b 100644
--- a/migration/rdma.c
+++ b/migration/rdma.c
@@ -4004,7 +4004,7 @@ static void rdma_accept_incoming_migration(void *opaque)
     }
 
     rdma->migration_started_on_destination = 1;
-    migration_fd_process_incoming(f);
+    migration_fd_process_incoming(f, errp);
 }
 
 void rdma_start_incoming_migration(const char *host_port, Error **errp)
-- 
2.23.0

[PATCH v2 06/10] migration: Add multifd-compress parameter

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>

---
Rename it to NONE
Fix typos (dave)
We don't need to chek values returned by visit_type_MultifdCompress (markus)
Fix yet more typos (wei)
---
 hw/core/qdev-properties.c    | 13 +++++++++++++
 include/hw/qdev-properties.h |  3 +++
 migration/migration.c        | 13 +++++++++++++
 monitor/hmp-cmds.c           | 13 +++++++++++++
 qapi/migration.json          | 30 +++++++++++++++++++++++++++---
 tests/migration-test.c       | 13 ++++++++++---
 6 files changed, 79 insertions(+), 6 deletions(-)

diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
index ac28890e5a..644705235e 100644
--- a/hw/core/qdev-properties.c
+++ b/hw/core/qdev-properties.c
@@ -8,6 +8,7 @@
 #include "qapi/qmp/qerror.h"
 #include "qemu/ctype.h"
 #include "qemu/error-report.h"
+#include "qapi/qapi-types-migration.h"
 #include "hw/block/block.h"
 #include "net/hub.h"
 #include "qapi/visitor.h"
@@ -648,6 +649,18 @@ const PropertyInfo qdev_prop_fdc_drive_type = {
     .set_default_value = set_default_value_enum,
 };
 
+/* --- MultifdCompress --- */
+
+const PropertyInfo qdev_prop_multifd_compress = {
+    .name = "MultifdCompress",
+    .description = "multifd_compress values, "
+                   "none",
+    .enum_table = &MultifdCompress_lookup,
+    .get = get_enum,
+    .set = set_enum,
+    .set_default_value = set_default_value_enum,
+};
+
 /* --- pci address --- */
 
 /*
diff --git a/include/hw/qdev-properties.h b/include/hw/qdev-properties.h
index c6a8cb5516..07d7bba682 100644
--- a/include/hw/qdev-properties.h
+++ b/include/hw/qdev-properties.h
@@ -21,6 +21,7 @@ extern const PropertyInfo qdev_prop_tpm;
 extern const PropertyInfo qdev_prop_ptr;
 extern const PropertyInfo qdev_prop_macaddr;
 extern const PropertyInfo qdev_prop_on_off_auto;
+extern const PropertyInfo qdev_prop_multifd_compress;
 extern const PropertyInfo qdev_prop_losttickpolicy;
 extern const PropertyInfo qdev_prop_blockdev_on_error;
 extern const PropertyInfo qdev_prop_bios_chs_trans;
@@ -204,6 +205,8 @@ extern const PropertyInfo qdev_prop_pcie_link_width;
     DEFINE_PROP(_n, _s, _f, qdev_prop_macaddr, MACAddr)
 #define DEFINE_PROP_ON_OFF_AUTO(_n, _s, _f, _d) \
     DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_on_off_auto, OnOffAuto)
+#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
+    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
 #define DEFINE_PROP_LOSTTICKPOLICY(_n, _s, _f, _d) \
     DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_losttickpolicy, \
                         LostTickPolicy)
diff --git a/migration/migration.c b/migration/migration.c
index cf6cec5fb6..93c6ed10a6 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -87,6 +87,7 @@
 /* The delay time (in ms) between two COLO checkpoints */
 #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY (200 * 100)
 #define DEFAULT_MIGRATE_MULTIFD_CHANNELS 16
+#define DEFAULT_MIGRATE_MULTIFD_COMPRESS MULTIFD_COMPRESS_NONE
 
 /* Background transfer rate for postcopy, 0 means unlimited, note
  * that page requests can still exceed this limit.
@@ -774,6 +775,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
     params->block_incremental = s->parameters.block_incremental;
     params->has_multifd_channels = true;
     params->multifd_channels = s->parameters.multifd_channels;
+    params->has_multifd_compress = true;
+    params->multifd_compress = s->parameters.multifd_compress;
     params->has_xbzrle_cache_size = true;
     params->xbzrle_cache_size = s->parameters.xbzrle_cache_size;
     params->has_max_postcopy_bandwidth = true;
@@ -1281,6 +1284,9 @@ static void migrate_params_test_apply(MigrateSetParameters *params,
     if (params->has_multifd_channels) {
         dest->multifd_channels = params->multifd_channels;
     }
+    if (params->has_multifd_compress) {
+        dest->multifd_compress = params->multifd_compress;
+    }
     if (params->has_xbzrle_cache_size) {
         dest->xbzrle_cache_size = params->xbzrle_cache_size;
     }
@@ -1377,6 +1383,9 @@ static void migrate_params_apply(MigrateSetParameters *params, Error **errp)
     if (params->has_multifd_channels) {
         s->parameters.multifd_channels = params->multifd_channels;
     }
+    if (params->has_multifd_compress) {
+        s->parameters.multifd_compress = params->multifd_compress;
+    }
     if (params->has_xbzrle_cache_size) {
         s->parameters.xbzrle_cache_size = params->xbzrle_cache_size;
         xbzrle_cache_resize(params->xbzrle_cache_size, errp);
@@ -3474,6 +3483,9 @@ static Property migration_properties[] = {
     DEFINE_PROP_UINT8("multifd-channels", MigrationState,
                       parameters.multifd_channels,
                       DEFAULT_MIGRATE_MULTIFD_CHANNELS),
+    DEFINE_PROP_MULTIFD_COMPRESS("multifd-compress", MigrationState,
+                      parameters.multifd_compress,
+                      DEFAULT_MIGRATE_MULTIFD_COMPRESS),
     DEFINE_PROP_SIZE("xbzrle-cache-size", MigrationState,
                       parameters.xbzrle_cache_size,
                       DEFAULT_MIGRATE_XBZRLE_CACHE_SIZE),
@@ -3564,6 +3576,7 @@ static void migration_instance_init(Object *obj)
     params->has_x_checkpoint_delay = true;
     params->has_block_incremental = true;
     params->has_multifd_channels = true;
+    params->has_multifd_compress = true;
     params->has_xbzrle_cache_size = true;
     params->has_max_postcopy_bandwidth = true;
     params->has_max_cpu_throttle = true;
diff --git a/monitor/hmp-cmds.c b/monitor/hmp-cmds.c
index b2551c16d1..caf06b0668 100644
--- a/monitor/hmp-cmds.c
+++ b/monitor/hmp-cmds.c
@@ -39,6 +39,7 @@
 #include "qapi/qapi-commands-tpm.h"
 #include "qapi/qapi-commands-ui.h"
 #include "qapi/qapi-visit-net.h"
+#include "qapi/qapi-visit-migration.h"
 #include "qapi/qmp/qdict.h"
 #include "qapi/qmp/qerror.h"
 #include "qapi/string-input-visitor.h"
@@ -448,6 +449,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
         monitor_printf(mon, "%s: %u\n",
             MigrationParameter_str(MIGRATION_PARAMETER_MULTIFD_CHANNELS),
             params->multifd_channels);
+        monitor_printf(mon, "%s: %s\n",
+            MigrationParameter_str(MIGRATION_PARAMETER_MULTIFD_COMPRESS),
+            MultifdCompress_str(params->multifd_compress));
         monitor_printf(mon, "%s: %" PRIu64 "\n",
             MigrationParameter_str(MIGRATION_PARAMETER_XBZRLE_CACHE_SIZE),
             params->xbzrle_cache_size);
@@ -1739,6 +1743,7 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
     MigrateSetParameters *p = g_new0(MigrateSetParameters, 1);
     uint64_t valuebw = 0;
     uint64_t cache_size;
+    MultifdCompress compress_type;
     Error *err = NULL;
     int val, ret;
 
@@ -1824,6 +1829,14 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
         p->has_multifd_channels = true;
         visit_type_int(v, param, &p->multifd_channels, &err);
         break;
+    case MIGRATION_PARAMETER_MULTIFD_COMPRESS:
+        p->has_multifd_compress = true;
+        visit_type_MultifdCompress(v, param, &compress_type, &err);
+        if (err) {
+            break;
+        }
+        p->multifd_compress = compress_type;
+        break;
     case MIGRATION_PARAMETER_XBZRLE_CACHE_SIZE:
         p->has_xbzrle_cache_size = true;
         visit_type_size(v, param, &cache_size, &err);
diff --git a/qapi/migration.json b/qapi/migration.json
index b7348d0c8b..430a39382e 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -488,6 +488,19 @@
 ##
 { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
 
+##
+# @MultifdCompress:
+#
+# An enumeration of multifd compression.
+#
+# @none: no compression.
+#
+# Since: 4.1
+#
+##
+{ 'enum': 'MultifdCompress',
+  'data': [ 'none' ] }
+
 ##
 # @MigrationParameter:
 #
@@ -586,6 +599,9 @@
 # @max-cpu-throttle: maximum cpu throttle percentage.
 #                    Defaults to 99. (Since 3.1)
 #
+# @multifd-compress: Which compression method to use.
+#                    Defaults to none. (Since 4.1)
+#
 # Since: 2.4
 ##
 { 'enum': 'MigrationParameter',
@@ -598,7 +614,7 @@
            'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
            'multifd-channels',
            'xbzrle-cache-size', 'max-postcopy-bandwidth',
-           'max-cpu-throttle' ] }
+           'max-cpu-throttle', 'multifd-compress' ] }
 
 ##
 # @MigrateSetParameters:
@@ -688,6 +704,9 @@
 # @max-cpu-throttle: maximum cpu throttle percentage.
 #                    The default value is 99. (Since 3.1)
 #
+# @multifd-compress: Which compression method to use.
+#                    Defaults to none. (Since 4.1)
+#
 # Since: 2.4
 ##
 # TODO either fuse back into MigrationParameters, or make
@@ -713,7 +732,8 @@
             '*multifd-channels': 'int',
             '*xbzrle-cache-size': 'size',
             '*max-postcopy-bandwidth': 'size',
-	    '*max-cpu-throttle': 'int' } }
+            '*max-cpu-throttle': 'int',
+            '*multifd-compress': 'MultifdCompress' } }
 
 ##
 # @migrate-set-parameters:
@@ -823,6 +843,9 @@
 #                    Defaults to 99.
 #                     (Since 3.1)
 #
+# @multifd-compress: Which compression method to use.
+#                    Defaults to none. (Since 4.1)
+#
 # Since: 2.4
 ##
 { 'struct': 'MigrationParameters',
@@ -846,7 +869,8 @@
             '*multifd-channels': 'uint8',
             '*xbzrle-cache-size': 'size',
 	    '*max-postcopy-bandwidth': 'size',
-            '*max-cpu-throttle':'uint8'} }
+            '*max-cpu-throttle': 'uint8',
+            '*multifd-compress': 'MultifdCompress' } }
 
 ##
 # @query-migrate-parameters:
diff --git a/tests/migration-test.c b/tests/migration-test.c
index fc221f172a..e5b6e54cfa 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -482,7 +482,6 @@ static void migrate_check_parameter_str(QTestState *who, const char *parameter,
     g_free(result);
 }
 
-__attribute__((unused))
 static void migrate_set_parameter_str(QTestState *who, const char *parameter,
                                       const char *value)
 {
@@ -1398,7 +1397,7 @@ static void test_migrate_auto_converge(void)
     test_migrate_end(from, to, true);
 }
 
-static void test_multifd_tcp(void)
+static void test_multifd_tcp(const char *method)
 {
     MigrateStart *args = migrate_start_new();
     QTestState *from, *to;
@@ -1422,6 +1421,9 @@ static void test_multifd_tcp(void)
     migrate_set_parameter_int(from, "multifd-channels", 16);
     migrate_set_parameter_int(to, "multifd-channels", 16);
 
+    migrate_set_parameter_str(from, "multifd-compress", method);
+    migrate_set_parameter_str(to, "multifd-compress", method);
+
     migrate_set_capability(from, "multifd", "true");
     migrate_set_capability(to, "multifd", "true");
 
@@ -1453,6 +1455,11 @@ static void test_multifd_tcp(void)
     free(uri);
 }
 
+static void test_multifd_tcp_none(void)
+{
+    test_multifd_tcp("none");
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1517,7 +1524,7 @@ int main(int argc, char **argv)
                    test_validate_uuid_dst_not_set);
 
     qtest_add_func("/migration/auto_converge", test_migrate_auto_converge);
-    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
+    qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
 
     ret = g_test_run();
 
-- 
2.23.0

[PATCH v2 07/10] migration: Make no compression operations into its own structure

[Juan Quintela]

It will be used later.

Signed-off-by: Juan Quintela <quintela@redhat.com>

---
Move setup of ->ops helper to proper place (wei)
Rename s/none/nocomp/ (dave)
Introduce MULTIFD_FLAG_NOCOMP
---
 migration/migration.c |   9 ++
 migration/migration.h |   1 +
 migration/ram.c       | 194 ++++++++++++++++++++++++++++++++++++++++--
 3 files changed, 196 insertions(+), 8 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index 93c6ed10a6..56203eb536 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -2213,6 +2213,15 @@ int migrate_multifd_channels(void)
     return s->parameters.multifd_channels;
 }
 
+int migrate_multifd_method(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->parameters.multifd_compress;
+}
+
 int migrate_use_xbzrle(void)
 {
     MigrationState *s;
diff --git a/migration/migration.h b/migration/migration.h
index 545f283ae7..d3ea45e25a 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -299,6 +299,7 @@ bool migrate_auto_converge(void);
 bool migrate_use_multifd(void);
 bool migrate_pause_before_switchover(void);
 int migrate_multifd_channels(void);
+int migrate_multifd_method(void);
 
 int migrate_use_xbzrle(void);
 int64_t migrate_xbzrle_cache_size(void);
diff --git a/migration/ram.c b/migration/ram.c
index fcf50e648a..10661e03ae 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -44,6 +44,7 @@
 #include "page_cache.h"
 #include "qemu/error-report.h"
 #include "qapi/error.h"
+#include "qapi/qapi-types-migration.h"
 #include "qapi/qapi-events-migration.h"
 #include "qapi/qmp/qerror.h"
 #include "trace.h"
@@ -581,6 +582,7 @@ exit:
 #define MULTIFD_VERSION 1
 
 #define MULTIFD_FLAG_SYNC (1 << 0)
+#define MULTIFD_FLAG_NOCOMP (1 << 1)
 
 /* This value needs to be a multiple of qemu_target_page_size() */
 #define MULTIFD_PACKET_SIZE (512 * 1024)
@@ -662,6 +664,8 @@ typedef struct {
     uint64_t num_pages;
     /* syncs main thread and channels */
     QemuSemaphore sem_sync;
+    /* used for compression methods */
+    void *data;
 }  MultiFDSendParams;
 
 typedef struct {
@@ -699,8 +703,153 @@ typedef struct {
     uint64_t num_pages;
     /* syncs main thread and channels */
     QemuSemaphore sem_sync;
+    /* used for de-compression methods */
+    void *data;
 } MultiFDRecvParams;
 
+typedef struct {
+    /* Setup for sending side */
+    int (*send_setup)(MultiFDSendParams *p, Error **errp);
+    /* Cleanup for sending side */
+    void (*send_cleanup)(MultiFDSendParams *p, Error **errp);
+    /* Prepare the send packet */
+    int (*send_prepare)(MultiFDSendParams *p, uint32_t used, Error **errp);
+    /* Write the send packet */
+    int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **errp);
+    /* Setup for receiving side */
+    int (*recv_setup)(MultiFDRecvParams *p, Error **errp);
+    /* Cleanup for receiving side */
+    void (*recv_cleanup)(MultiFDRecvParams *p);
+    /* Read all pages */
+    int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **errp);
+} MultiFDMethods;
+
+/* Multifd without compression */
+
+/**
+ * nocomp_send_setup: setup send side
+ *
+ * For no compression this function does nothing.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
+{
+    return 0;
+}
+
+/**
+ * nocomp_send_cleanup: cleanup send side
+ *
+ * For no compression this function does nothing.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
+{
+    return;
+}
+
+/**
+ * nocomp_send_prepare: prepare date to be able to send
+ *
+ * For no compression we just have to calculate the size of the
+ * packet.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used,
+                               Error **errp)
+{
+    p->next_packet_size = used * qemu_target_page_size();
+    p->flags |= MULTIFD_FLAG_NOCOMP;
+    return 0;
+}
+
+/**
+ * nocomp_send_write: do the actual write of the data
+ *
+ * For no compression we just have to write the data.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int nocomp_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
+{
+    return qio_channel_writev_all(p->c, p->pages->iov, used, errp);
+}
+
+/**
+ * nocomp_recv_setup: setup receive side
+ *
+ * For no compression this function does nothing.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
+{
+    return 0;
+}
+
+/**
+ * nocomp_recv_cleanup: setup receive side
+ *
+ * For no compression this function does nothing.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void nocomp_recv_cleanup(MultiFDRecvParams *p)
+{
+}
+
+/**
+ * nocomp_recv_pages: read the data from the channel into actual pages
+ *
+ * For no compression we just need to read things into the correct place.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
+{
+    if (p->flags != MULTIFD_FLAG_NOCOMP) {
+        error_setg(errp, "multifd %d: flags received %x flags expected %x",
+                   p->id, MULTIFD_FLAG_NOCOMP, p->flags);
+        return -1;
+    }
+    return qio_channel_readv_all(p->c, p->pages->iov, used, errp);
+}
+
+static MultiFDMethods multifd_nocomp_ops = {
+    .send_setup = nocomp_send_setup,
+    .send_cleanup = nocomp_send_cleanup,
+    .send_prepare = nocomp_send_prepare,
+    .send_write = nocomp_send_write,
+    .recv_setup = nocomp_recv_setup,
+    .recv_cleanup = nocomp_recv_cleanup,
+    .recv_pages = nocomp_recv_pages
+};
+
+static MultiFDMethods *multifd_ops[MULTIFD_COMPRESS__MAX] = {
+    [MULTIFD_COMPRESS_NONE] = &multifd_nocomp_ops,
+};
+
 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
 {
     MultiFDInit_t msg;
@@ -898,6 +1047,8 @@ struct {
     uint64_t packet_num;
     /* send channels ready */
     QemuSemaphore channels_ready;
+    /* multifd ops */
+    MultiFDMethods *ops;
 } *multifd_send_state;
 
 /*
@@ -1027,6 +1178,7 @@ void multifd_save_cleanup(void)
     multifd_send_terminate_threads(NULL);
     for (i = 0; i < migrate_multifd_channels(); i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
+        Error *local_err = NULL;
 
         if (p->running) {
             qemu_thread_join(&p->thread);
@@ -1043,6 +1195,10 @@ void multifd_save_cleanup(void)
         p->packet_len = 0;
         g_free(p->packet);
         p->packet = NULL;
+        multifd_send_state->ops->send_cleanup(p, &local_err);
+        if (local_err) {
+            migrate_set_error(migrate_get_current(), local_err);
+        }
     }
     qemu_sem_destroy(&multifd_send_state->channels_ready);
     g_free(multifd_send_state->params);
@@ -1123,7 +1279,14 @@ static void *multifd_send_thread(void *opaque)
             uint64_t packet_num = p->packet_num;
             flags = p->flags;
 
-            p->next_packet_size = used * qemu_target_page_size();
+            if (used) {
+                ret = multifd_send_state->ops->send_prepare(p, used,
+                                                            &local_err);
+                if (ret != 0) {
+                    qemu_mutex_unlock(&p->mutex);
+                    break;
+                }
+            }
             multifd_send_fill_packet(p);
             p->flags = 0;
             p->num_packets++;
@@ -1140,8 +1303,7 @@ static void *multifd_send_thread(void *opaque)
             }
 
             if (used) {
-                ret = qio_channel_writev_all(p->c, p->pages->iov,
-                                             used, &local_err);
+                ret = multifd_send_state->ops->send_write(p, used, &local_err);
                 if (ret != 0) {
                     break;
                 }
@@ -1212,6 +1374,7 @@ int multifd_save_setup(Error **errp)
 {
     int thread_count;
     uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    int ret = 0;
     uint8_t i;
 
     if (!migrate_use_multifd()) {
@@ -1222,9 +1385,11 @@ int multifd_save_setup(Error **errp)
     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
     multifd_send_state->pages = multifd_pages_init(page_count);
     qemu_sem_init(&multifd_send_state->channels_ready, 0);
+    multifd_send_state->ops = multifd_ops[migrate_multifd_method()];
 
     for (i = 0; i < thread_count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
+        int res;
 
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem, 0);
@@ -1240,8 +1405,12 @@ int multifd_save_setup(Error **errp)
         p->packet->version = cpu_to_be32(MULTIFD_VERSION);
         p->name = g_strdup_printf("multifdsend_%d", i);
         socket_send_channel_create(multifd_new_send_channel_async, p);
+        res = multifd_send_state->ops->send_setup(p, errp);
+        if (ret == 0) {
+            ret = res;
+        }
     }
-    return 0;
+    return ret;
 }
 
 struct {
@@ -1252,6 +1421,8 @@ struct {
     QemuSemaphore sem_sync;
     /* global number of generated multifd packets */
     uint64_t packet_num;
+    /* multifd ops */
+    MultiFDMethods *ops;
 } *multifd_recv_state;
 
 static void multifd_recv_terminate_threads(Error *err)
@@ -1287,7 +1458,6 @@ static void multifd_recv_terminate_threads(Error *err)
 int multifd_load_cleanup(Error **errp)
 {
     int i;
-    int ret = 0;
 
     if (!migrate_use_multifd()) {
         return 0;
@@ -1316,6 +1486,7 @@ int multifd_load_cleanup(Error **errp)
         p->packet_len = 0;
         g_free(p->packet);
         p->packet = NULL;
+        multifd_recv_state->ops->recv_cleanup(p);
     }
     qemu_sem_destroy(&multifd_recv_state->sem_sync);
     g_free(multifd_recv_state->params);
@@ -1323,7 +1494,7 @@ int multifd_load_cleanup(Error **errp)
     g_free(multifd_recv_state);
     multifd_recv_state = NULL;
 
-    return ret;
+    return 0;
 }
 
 static void multifd_recv_sync_main(void)
@@ -1388,6 +1559,8 @@ static void *multifd_recv_thread(void *opaque)
 
         used = p->pages->used;
         flags = p->flags;
+        /* recv methods don't know how to handle the SYNC flag */
+        p->flags &= ~MULTIFD_FLAG_SYNC;
         trace_multifd_recv(p->id, p->packet_num, used, flags,
                            p->next_packet_size);
         p->num_packets++;
@@ -1395,8 +1568,7 @@ static void *multifd_recv_thread(void *opaque)
         qemu_mutex_unlock(&p->mutex);
 
         if (used) {
-            ret = qio_channel_readv_all(p->c, p->pages->iov,
-                                        used, &local_err);
+            ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
             if (ret != 0) {
                 break;
             }
@@ -1435,9 +1607,11 @@ int multifd_load_setup(Error **errp)
     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
     atomic_set(&multifd_recv_state->count, 0);
     qemu_sem_init(&multifd_recv_state->sem_sync, 0);
+    multifd_recv_state->ops = multifd_ops[migrate_multifd_method()];
 
     for (i = 0; i < thread_count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
+        int ret;
 
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem_sync, 0);
@@ -1448,6 +1622,10 @@ int multifd_load_setup(Error **errp)
                       + sizeof(ram_addr_t) * page_count;
         p->packet = g_malloc0(p->packet_len);
         p->name = g_strdup_printf("multifdrecv_%d", i);
+        ret = multifd_recv_state->ops->recv_setup(p, errp);
+        if (ret != 0) {
+            return ret;
+        }
     }
     return 0;
 }
-- 
2.23.0

[PATCH v2 08/10] migration: Add zlib compression multifd support

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 hw/core/qdev-properties.c |   2 +-
 migration/ram.c           | 264 ++++++++++++++++++++++++++++++++++++++
 qapi/migration.json       |   2 +-
 tests/migration-test.c    |   6 +
 4 files changed, 272 insertions(+), 2 deletions(-)

diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
index 644705235e..e8ff317a60 100644
--- a/hw/core/qdev-properties.c
+++ b/hw/core/qdev-properties.c
@@ -654,7 +654,7 @@ const PropertyInfo qdev_prop_fdc_drive_type = {
 const PropertyInfo qdev_prop_multifd_compress = {
     .name = "MultifdCompress",
     .description = "multifd_compress values, "
-                   "none",
+                   "none/zlib",
     .enum_table = &MultifdCompress_lookup,
     .get = get_enum,
     .set = set_enum,
diff --git a/migration/ram.c b/migration/ram.c
index 10661e03ae..5006d719b4 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -583,6 +583,7 @@ exit:
 
 #define MULTIFD_FLAG_SYNC (1 << 0)
 #define MULTIFD_FLAG_NOCOMP (1 << 1)
+#define MULTIFD_FLAG_ZLIB (1 << 2)
 
 /* This value needs to be a multiple of qemu_target_page_size() */
 #define MULTIFD_PACKET_SIZE (512 * 1024)
@@ -625,6 +626,15 @@ typedef struct {
     RAMBlock *block;
 } MultiFDPages_t;
 
+struct zlib_data {
+    /* stream for compression */
+    z_stream zs;
+    /* compressed buffer */
+    uint8_t *zbuff;
+    /* size of compressed buffer */
+    uint32_t zbuff_len;
+};
+
 typedef struct {
     /* this fields are not changed once the thread is created */
     /* channel number */
@@ -846,8 +856,262 @@ static MultiFDMethods multifd_nocomp_ops = {
     .recv_pages = nocomp_recv_pages
 };
 
+/* Multifd zlib compression */
+
+/**
+ * zlib_send_setup: setup send side
+ *
+ * Setup each channel with zlib compression.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int zlib_send_setup(MultiFDSendParams *p, Error **errp)
+{
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    struct zlib_data *z = g_malloc0(sizeof(struct zlib_data));
+    z_stream *zs = &z->zs;
+
+    p->data = z;
+    zs->zalloc = Z_NULL;
+    zs->zfree = Z_NULL;
+    zs->opaque = Z_NULL;
+    if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
+        g_free(z);
+        error_setg(errp, "multifd %d: deflate init failed", p->id);
+        return -1;
+    }
+    /* We will never have more than page_count pages */
+    z->zbuff_len = page_count * qemu_target_page_size();
+    z->zbuff_len *= 2;
+    z->zbuff = g_try_malloc(z->zbuff_len);
+    if (!z->zbuff) {
+        g_free(z);
+        error_setg(errp, "multifd %d: out of memory for zbuff", p->id);
+        return -1;
+    }
+    return 0;
+}
+
+/**
+ * zlib_send_cleanup: cleanup send side
+ *
+ * Close the channel and return memory.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void zlib_send_cleanup(MultiFDSendParams *p, Error **errp)
+{
+    struct zlib_data *z = p->data;
+
+    deflateEnd(&z->zs);
+    g_free(z->zbuff);
+    z->zbuff = NULL;
+    g_free(p->data);
+    p->data = NULL;
+}
+
+/**
+ * zlib_send_prepare: prepare date to be able to send
+ *
+ * Create a compressed buffer with all the pages that we are going to
+ * send.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ */
+static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used, Error **errp)
+{
+    struct iovec *iov = p->pages->iov;
+    struct zlib_data *z = p->data;
+    z_stream *zs = &z->zs;
+    uint32_t out_size = 0;
+    int ret;
+    uint32_t i;
+
+    for (i = 0; i < used; i++) {
+        uint32_t available = z->zbuff_len - out_size;
+        int flush = Z_NO_FLUSH;
+
+        if (i == used  - 1) {
+            flush = Z_SYNC_FLUSH;
+        }
+
+        zs->avail_in = iov[i].iov_len;
+        zs->next_in = iov[i].iov_base;
+
+        zs->avail_out = available;
+        zs->next_out = z->zbuff + out_size;
+
+        ret = deflate(zs, flush);
+        if (ret != Z_OK) {
+            error_setg(errp, "multifd %d: deflate returned %d instead of Z_OK",
+                       p->id, ret);
+            return -1;
+        }
+        out_size += available - zs->avail_out;
+    }
+    p->next_packet_size = out_size;
+    p->flags |= MULTIFD_FLAG_ZLIB;
+
+    return 0;
+}
+
+/**
+ * zlib_send_write: do the actual write of the data
+ *
+ * Do the actual write of the comprresed buffer.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int zlib_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
+{
+    struct zlib_data *z = p->data;
+
+    return qio_channel_write_all(p->c, (void *)z->zbuff, p->next_packet_size,
+                                 errp);
+}
+
+/**
+ * zlib_recv_setup: setup receive side
+ *
+ * Create the compressed channel and buffer.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int zlib_recv_setup(MultiFDRecvParams *p, Error **errp)
+{
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    struct zlib_data *z = g_malloc0(sizeof(struct zlib_data));
+    z_stream *zs = &z->zs;
+
+    p->data = z;
+    zs->zalloc = Z_NULL;
+    zs->zfree = Z_NULL;
+    zs->opaque = Z_NULL;
+    zs->avail_in = 0;
+    zs->next_in = Z_NULL;
+    if (inflateInit(zs) != Z_OK) {
+        error_setg(errp, "multifd %d: inflate init failed", p->id);
+        return -1;
+    }
+    /* We will never have more than page_count pages */
+    z->zbuff_len = page_count * qemu_target_page_size();
+    /* We know compression "could" use more space */
+    z->zbuff_len *= 2;
+    z->zbuff = g_try_malloc(z->zbuff_len);
+    if (!z->zbuff) {
+        error_setg(errp, "multifd %d: out of memory for zbuff", p->id);
+        return -1;
+    }
+    return 0;
+}
+
+/**
+ * zlib_recv_cleanup: setup receive side
+ *
+ * For no compression this function does nothing.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void zlib_recv_cleanup(MultiFDRecvParams *p)
+{
+    struct zlib_data *z = p->data;
+
+    inflateEnd(&z->zs);
+    g_free(z->zbuff);
+    z->zbuff = NULL;
+    g_free(p->data);
+    p->data = NULL;
+}
+
+/**
+ * zlib_recv_pages: read the data from the channel into actual pages
+ *
+ * Read the compressed buffer, and uncompress it into the actual
+ * pages.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
+{
+    uint32_t in_size = p->next_packet_size;
+    uint32_t out_size = 0;
+    uint32_t expected_size = used * qemu_target_page_size();
+    struct zlib_data *z = p->data;
+    z_stream *zs = &z->zs;
+    int ret;
+    int i;
+
+    if (p->flags != MULTIFD_FLAG_ZLIB) {
+        error_setg(errp, "multifd %d: flags received %x flags expected %x",
+                   p->id, MULTIFD_FLAG_ZLIB, p->flags);
+        return -1;
+    }
+    ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
+
+    if (ret != 0) {
+        return ret;
+    }
+
+    zs->avail_in = in_size;
+    zs->next_in = z->zbuff;
+
+    for (i = 0; i < used; i++) {
+        struct iovec *iov = &p->pages->iov[i];
+        int flush = Z_NO_FLUSH;
+
+        if (i == used  - 1) {
+            flush = Z_SYNC_FLUSH;
+        }
+
+        zs->avail_out = iov->iov_len;
+        zs->next_out = iov->iov_base;
+
+        ret = inflate(zs, flush);
+        if (ret != Z_OK) {
+            error_setg(errp, "multifd %d: inflate returned %d instead of Z_OK",
+                       p->id, ret);
+            return ret;
+        }
+        out_size += iov->iov_len;
+    }
+    if (out_size != expected_size) {
+        error_setg(errp, "multifd %d: packet size received %d size expected %d",
+                   p->id, out_size, expected_size);
+        return -1;
+    }
+    return 0;
+}
+
+static MultiFDMethods multifd_zlib_ops = {
+    .send_setup = zlib_send_setup,
+    .send_cleanup = zlib_send_cleanup,
+    .send_prepare = zlib_send_prepare,
+    .send_write = zlib_send_write,
+    .recv_setup = zlib_recv_setup,
+    .recv_cleanup = zlib_recv_cleanup,
+    .recv_pages = zlib_recv_pages
+};
+
 static MultiFDMethods *multifd_ops[MULTIFD_COMPRESS__MAX] = {
     [MULTIFD_COMPRESS_NONE] = &multifd_nocomp_ops,
+    [MULTIFD_COMPRESS_ZLIB] = &multifd_zlib_ops,
 };
 
 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
diff --git a/qapi/migration.json b/qapi/migration.json
index 430a39382e..0e3a3db8d0 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -499,7 +499,7 @@
 #
 ##
 { 'enum': 'MultifdCompress',
-  'data': [ 'none' ] }
+  'data': [ 'none', 'zlib' ] }
 
 ##
 # @MigrationParameter:
diff --git a/tests/migration-test.c b/tests/migration-test.c
index e5b6e54cfa..8985e05c69 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -1460,6 +1460,11 @@ static void test_multifd_tcp_none(void)
     test_multifd_tcp("none");
 }
 
+static void test_multifd_tcp_zlib(void)
+{
+    test_multifd_tcp("zlib");
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1525,6 +1530,7 @@ int main(int argc, char **argv)
 
     qtest_add_func("/migration/auto_converge", test_migrate_auto_converge);
     qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
+    qtest_add_func("/migration/multifd/tcp/zlib", test_multifd_tcp_zlib);
 
     ret = g_test_run();
 
-- 
2.23.0

[PATCH v2 09/10] configure: Enable test and libs for zstd

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 configure | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/configure b/configure
index 84b413dbfc..a8f3027c67 100755
--- a/configure
+++ b/configure
@@ -447,6 +447,7 @@ lzo=""
 snappy=""
 bzip2=""
 lzfse=""
+zstd=""
 guest_agent=""
 guest_agent_with_vss="no"
 guest_agent_ntddscsi="no"
@@ -1341,6 +1342,10 @@ for opt do
   ;;
   --disable-lzfse) lzfse="no"
   ;;
+  --disable-zstd) zstd="no"
+  ;;
+  --enable-zstd) zstd="yes"
+  ;;
   --enable-guest-agent) guest_agent="yes"
   ;;
   --disable-guest-agent) guest_agent="no"
@@ -1788,6 +1793,8 @@ disabled with --disable-FEATURE, default is enabled if available:
                   (for reading bzip2-compressed dmg images)
   lzfse           support of lzfse compression library
                   (for reading lzfse-compressed dmg images)
+  zstd            support for zstd compression library
+                  (for migration compression)
   seccomp         seccomp support
   coroutine-pool  coroutine freelist (better performance)
   glusterfs       GlusterFS backend
@@ -2401,6 +2408,24 @@ EOF
     fi
 fi
 
+##########################################
+# zstd check
+
+if test "$zstd" != "no" ; then
+    if $pkg_config --exist libzstd ; then
+        zstd_cflags="$($pkg_config --cflags libzstd)"
+        zstd_libs="$($pkg_config --libs libzstd)"
+        LIBS="$zstd_libs $LIBS"
+        QEMU_CFLAGS="$QEMU_CFLAGS $zstd_cflags"
+        zstd="yes"
+    else
+        if test "$zstd" = "yes" ; then
+            feature_not_found "libzstd" "Install libzstd devel"
+        fi
+        zstd="no"
+    fi
+fi
+
 ##########################################
 # libseccomp check
 
@@ -6535,6 +6560,7 @@ echo "lzo support       $lzo"
 echo "snappy support    $snappy"
 echo "bzip2 support     $bzip2"
 echo "lzfse support     $lzfse"
+echo "zstd support      $zstd"
 echo "NUMA host support $numa"
 echo "libxml2           $libxml2"
 echo "tcmalloc support  $tcmalloc"
@@ -7102,6 +7128,10 @@ if test "$lzfse" = "yes" ; then
   echo "LZFSE_LIBS=-llzfse" >> $config_host_mak
 fi
 
+if test "$zstd" = "yes" ; then
+  echo "CONFIG_ZSTD=y" >> $config_host_mak
+fi
+
 if test "$libiscsi" = "yes" ; then
   echo "CONFIG_LIBISCSI=m" >> $config_host_mak
   echo "LIBISCSI_CFLAGS=$libiscsi_cflags" >> $config_host_mak
-- 
2.23.0

[PATCH v2 10/10] migration: Add zstd compression multifd support

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 hw/core/qdev-properties.c |   2 +-
 migration/ram.c           | 288 ++++++++++++++++++++++++++++++++++++++
 qapi/migration.json       |   2 +-
 tests/migration-test.c    |   6 +
 4 files changed, 296 insertions(+), 2 deletions(-)

diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
index e8ff317a60..b75467704f 100644
--- a/hw/core/qdev-properties.c
+++ b/hw/core/qdev-properties.c
@@ -654,7 +654,7 @@ const PropertyInfo qdev_prop_fdc_drive_type = {
 const PropertyInfo qdev_prop_multifd_compress = {
     .name = "MultifdCompress",
     .description = "multifd_compress values, "
-                   "none/zlib",
+                   "none/zlib/zstd",
     .enum_table = &MultifdCompress_lookup,
     .get = get_enum,
     .set = set_enum,
diff --git a/migration/ram.c b/migration/ram.c
index 5006d719b4..db90237977 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -29,6 +29,9 @@
 #include "qemu/osdep.h"
 #include "cpu.h"
 #include <zlib.h>
+#ifdef CONFIG_ZSTD
+#include <zstd.h>
+#endif
 #include "qemu/cutils.h"
 #include "qemu/bitops.h"
 #include "qemu/bitmap.h"
@@ -584,6 +587,7 @@ exit:
 #define MULTIFD_FLAG_SYNC (1 << 0)
 #define MULTIFD_FLAG_NOCOMP (1 << 1)
 #define MULTIFD_FLAG_ZLIB (1 << 2)
+#define MULTIFD_FLAG_ZSTD (1 << 3)
 
 /* This value needs to be a multiple of qemu_target_page_size() */
 #define MULTIFD_PACKET_SIZE (512 * 1024)
@@ -635,6 +639,22 @@ struct zlib_data {
     uint32_t zbuff_len;
 };
 
+#ifdef CONFIG_ZSTD
+struct zstd_data {
+    /* stream for compression */
+    ZSTD_CStream *zcs;
+    /* stream for decompression */
+    ZSTD_DStream *zds;
+    /* buffers */
+    ZSTD_inBuffer in;
+    ZSTD_outBuffer out;
+    /* compressed buffer */
+    uint8_t *zbuff;
+    /* size of compressed buffer */
+    uint32_t zbuff_len;
+};
+#endif
+
 typedef struct {
     /* this fields are not changed once the thread is created */
     /* channel number */
@@ -1109,9 +1129,277 @@ static MultiFDMethods multifd_zlib_ops = {
     .recv_pages = zlib_recv_pages
 };
 
+#ifdef CONFIG_ZSTD
+/* Multifd zstd compression */
+
+/**
+ * zstd_send_setup: setup send side
+ *
+ * Setup each channel with zstd compression.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int zstd_send_setup(MultiFDSendParams *p, Error **errp)
+{
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    struct zstd_data *z = g_new0(struct zstd_data, 1);
+    int res;
+
+    p->data = z;
+    z->zcs = ZSTD_createCStream();
+    if (!z->zcs) {
+        g_free(z);
+        error_setg(errp, "multifd %d: zstd createCStream failed", p->id);
+        return -1;
+    }
+
+    res = ZSTD_initCStream(z->zcs, migrate_compress_level());
+    if (ZSTD_isError(res)) {
+        ZSTD_freeCStream(z->zcs);
+        g_free(z);
+        error_setg(errp, "multifd %d: initCStream failed", p->id);
+        return -1;
+    }
+    /* We will never have more than page_count pages */
+    z->zbuff_len = page_count * qemu_target_page_size();
+    z->zbuff_len *= 2;
+    z->zbuff = g_try_malloc(z->zbuff_len);
+    if (!z->zbuff) {
+        ZSTD_freeCStream(z->zcs);
+        g_free(z);
+        error_setg(errp, "multifd %d: out of memory for zbuff", p->id);
+        return -1;
+    }
+    return 0;
+}
+
+/**
+ * zstd_send_cleanup: cleanup send side
+ *
+ * Close the channel and return memory.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void zstd_send_cleanup(MultiFDSendParams *p, Error **errp)
+{
+    struct zstd_data *z = p->data;
+
+    ZSTD_freeCStream(z->zcs);
+    z->zcs = NULL;
+    g_free(z->zbuff);
+    z->zbuff = NULL;
+    g_free(p->data);
+    p->data = NULL;
+}
+
+/**
+ * zstd_send_prepare: prepare date to be able to send
+ *
+ * Create a compressed buffer with all the pages that we are going to
+ * send.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ */
+static int zstd_send_prepare(MultiFDSendParams *p, uint32_t used, Error **errp)
+{
+    struct iovec *iov = p->pages->iov;
+    struct zstd_data *z = p->data;
+    int ret;
+    uint32_t i;
+
+    z->out.dst = z->zbuff;
+    z->out.size = z->zbuff_len;
+    z->out.pos = 0;
+
+    for (i = 0; i < used; i++) {
+        ZSTD_EndDirective flush = ZSTD_e_continue;
+
+        if (i == used - 1) {
+            flush = ZSTD_e_flush;
+        }
+        z->in.src = iov[i].iov_base;
+        z->in.size = iov[i].iov_len;
+        z->in.pos = 0;
+
+        ret = ZSTD_compressStream2(z->zcs, &z->out, &z->in, flush);
+        if (ZSTD_isError(ret)) {
+            error_setg(errp, "multifd %d: compressStream error %s",
+                       p->id, ZSTD_getErrorName(ret));
+            return -1;
+        }
+    }
+    p->next_packet_size = z->out.pos;
+    p->flags |= MULTIFD_FLAG_ZSTD;
+
+    return 0;
+}
+
+/**
+ * zstd_send_write: do the actual write of the data
+ *
+ * Do the actual write of the comprresed buffer.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int zstd_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
+{
+    struct zstd_data *z = p->data;
+
+    return qio_channel_write_all(p->c, (void *)z->zbuff, p->next_packet_size,
+                                 errp);
+}
+
+/**
+ * zstd_recv_setup: setup receive side
+ *
+ * Create the compressed channel and buffer.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @errp: pointer to an error
+ */
+static int zstd_recv_setup(MultiFDRecvParams *p, Error **errp)
+{
+    uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
+    struct zstd_data *z = g_new0(struct zstd_data, 1);
+    int res;
+
+    p->data = z;
+    z->zds = ZSTD_createDStream();
+    if (!z->zds) {
+        g_free(z);
+        error_setg(errp, "multifd %d: zstd createDStream failed", p->id);
+        return -1;
+    }
+
+    res = ZSTD_initDStream(z->zds);
+    if (ZSTD_isError(res)) {
+        ZSTD_freeDStream(z->zds);
+        g_free(z);
+        error_setg(errp, "multifd %d: initDStream failed", p->id);
+        return -1;
+    }
+
+    /* We will never have more than page_count pages */
+    z->zbuff_len = page_count * qemu_target_page_size();
+    /* We know compression "could" use more space */
+    z->zbuff_len *= 2;
+    z->zbuff = g_try_malloc(z->zbuff_len);
+    if (!z->zbuff) {
+        ZSTD_freeDStream(z->zds);
+        g_free(z);
+        error_setg(errp, "multifd %d: out of memory for zbuff", p->id);
+        return -1;
+    }
+    return 0;
+}
+
+/**
+ * zstd_recv_cleanup: setup receive side
+ *
+ * For no compression this function does nothing.
+ *
+ * @p: Params for the channel that we are using
+ */
+static void zstd_recv_cleanup(MultiFDRecvParams *p)
+{
+    struct zstd_data *z = p->data;
+
+    ZSTD_freeDStream(z->zds);
+    z->zds = NULL;
+    g_free(z->zbuff);
+    z->zbuff = NULL;
+    g_free(p->data);
+    p->data = NULL;
+}
+
+/**
+ * zstd_recv_pages: read the data from the channel into actual pages
+ *
+ * Read the compressed buffer, and uncompress it into the actual
+ * pages.
+ *
+ * Returns 0 for success or -1 for error
+ *
+ * @p: Params for the channel that we are using
+ * @used: number of pages used
+ * @errp: pointer to an error
+ */
+static int zstd_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
+{
+    uint32_t in_size = p->next_packet_size;
+    uint32_t out_size = 0;
+    uint32_t expected_size = used * qemu_target_page_size();
+    struct zstd_data *z = p->data;
+    int ret;
+    int i;
+
+    if (p->flags != MULTIFD_FLAG_ZSTD) {
+        error_setg(errp, "multifd %d: flags received %x flags expected %x",
+                   p->id, MULTIFD_FLAG_ZSTD, p->flags);
+        return -1;
+    }
+    ret = qio_channel_read_all(p->c, (void *)z->zbuff, in_size, errp);
+
+    if (ret != 0) {
+        return ret;
+    }
+
+    z->in.src = z->zbuff;
+    z->in.size = in_size;
+    z->in.pos = 0;
+
+    for (i = 0; i < used; i++) {
+        struct iovec *iov = &p->pages->iov[i];
+
+        z->out.dst = iov->iov_base;
+        z->out.size = iov->iov_len;
+        z->out.pos = 0;
+
+        ret = ZSTD_decompressStream(z->zds, &z->out, &z->in);
+        if (ZSTD_isError(ret)) {
+            error_setg(errp, "multifd %d: decompressStream returned %s",
+                       p->id, ZSTD_getErrorName(ret));
+            return ret;
+        }
+        out_size += iov->iov_len;
+    }
+    if (out_size != expected_size) {
+        error_setg(errp, "multifd %d: packet size received %d size expected %d",
+                   p->id, out_size, expected_size);
+        return -1;
+    }
+    return 0;
+}
+
+static MultiFDMethods multifd_zstd_ops = {
+    .send_setup = zstd_send_setup,
+    .send_cleanup = zstd_send_cleanup,
+    .send_prepare = zstd_send_prepare,
+    .send_write = zstd_send_write,
+    .recv_setup = zstd_recv_setup,
+    .recv_cleanup = zstd_recv_cleanup,
+    .recv_pages = zstd_recv_pages
+};
+#endif /* CONFIG_ZSTD */
+
 static MultiFDMethods *multifd_ops[MULTIFD_COMPRESS__MAX] = {
     [MULTIFD_COMPRESS_NONE] = &multifd_nocomp_ops,
     [MULTIFD_COMPRESS_ZLIB] = &multifd_zlib_ops,
+#ifdef CONFIG_ZSTD
+    [MULTIFD_COMPRESS_ZSTD] = &multifd_zstd_ops,
+#endif
 };
 
 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
diff --git a/qapi/migration.json b/qapi/migration.json
index 0e3a3db8d0..1b827e55b6 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -499,7 +499,7 @@
 #
 ##
 { 'enum': 'MultifdCompress',
-  'data': [ 'none', 'zlib' ] }
+  'data': [ 'none', 'zlib', 'zstd' ] }
 
 ##
 # @MigrationParameter:
diff --git a/tests/migration-test.c b/tests/migration-test.c
index 8985e05c69..7588f50b9b 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -1465,6 +1465,11 @@ static void test_multifd_tcp_zlib(void)
     test_multifd_tcp("zlib");
 }
 
+static void test_multifd_tcp_zstd(void)
+{
+    test_multifd_tcp("zstd");
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1531,6 +1536,7 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/auto_converge", test_migrate_auto_converge);
     qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
     qtest_add_func("/migration/multifd/tcp/zlib", test_multifd_tcp_zlib);
+    qtest_add_func("/migration/multifd/tcp/zstd", test_multifd_tcp_zstd);
 
     ret = g_test_run();
 
-- 
2.23.0

Re: [PATCH v2 06/10] migration: Add multifd-compress parameter

[Markus Armbruster]

Juan Quintela <quintela@redhat.com> writes:

> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> ---
> Rename it to NONE
> Fix typos (dave)
> We don't need to chek values returned by visit_type_MultifdCompress (markus)
> Fix yet more typos (wei)
> ---
>  hw/core/qdev-properties.c    | 13 +++++++++++++
>  include/hw/qdev-properties.h |  3 +++
>  migration/migration.c        | 13 +++++++++++++
>  monitor/hmp-cmds.c           | 13 +++++++++++++
>  qapi/migration.json          | 30 +++++++++++++++++++++++++++---
>  tests/migration-test.c       | 13 ++++++++++---
>  6 files changed, 79 insertions(+), 6 deletions(-)

For QAPI:
Acked-by: Markus Armbruster <armbru@redhat.com>

Re: [PATCH v2 04/10] migration: Make multifd_save_setup() get an Error parameter

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/migration.c | 2 +-
>  migration/ram.c       | 2 +-
>  migration/ram.h       | 2 +-
>  3 files changed, 3 insertions(+), 3 deletions(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index e7f707e033..5a56bd0c91 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -3400,7 +3400,7 @@ void migrate_fd_connect(MigrationState *s, Error *error_in)
>          return;
>      }
>  
> -    if (multifd_save_setup() != 0) {
> +    if (multifd_save_setup(&error_in) != 0) {

I'm not sure that's right.  I think the *error passed into
migration_channel_connect, and then onto migrate_fd_connect is an
indication that an error has happened, not a place you can put
an error pointer.   Note how migration_channel_connect
frees it after the migrate_fd_connect call, it doesn't report it.

Dave

>          migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
>                            MIGRATION_STATUS_FAILED);
>          migrate_fd_cleanup(s);
> diff --git a/migration/ram.c b/migration/ram.c
> index 38070f1bb2..1f364cc23d 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -1208,7 +1208,7 @@ static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
>      }
>  }
>  
> -int multifd_save_setup(void)
> +int multifd_save_setup(Error **errp)
>  {
>      int thread_count;
>      uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> diff --git a/migration/ram.h b/migration/ram.h
> index bd0eee79b6..da22a417ea 100644
> --- a/migration/ram.h
> +++ b/migration/ram.h
> @@ -41,7 +41,7 @@ int xbzrle_cache_resize(int64_t new_size, Error **errp);
>  uint64_t ram_bytes_remaining(void);
>  uint64_t ram_bytes_total(void);
>  
> -int multifd_save_setup(void);
> +int multifd_save_setup(Error **errp);
>  void multifd_save_cleanup(void);
>  int multifd_load_setup(void);
>  int multifd_load_cleanup(Error **errp);
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> We can scale much better with 16, so we can scale to higher numbers.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/migration.c | 2 +-
>  1 file changed, 1 insertion(+), 1 deletion(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index 354ad072fa..e7f707e033 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -86,7 +86,7 @@
>  
>  /* The delay time (in ms) between two COLO checkpoints */
>  #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY (200 * 100)
> -#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 2
> +#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 16

OK, I suspect the management apps will set it anyway; some could be a
bit surprised chewing 16 cores/threads of CPU in some cases.


Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

>  /* Background transfer rate for postcopy, 0 means unlimited, note
>   * that page requests can still exceed this limit.
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 03/10] migration-test: introduce functions to handle string parameters

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>

Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

(I'm surprised we don't need to quote the %s's in qtest_qmp, but it
seems that we never do it in any other of the qtest_qmp calls that I can
see when we use %s.

Dave

> ---
>  tests/migration-test.c | 37 +++++++++++++++++++++++++++++++++++++
>  1 file changed, 37 insertions(+)
> 
> diff --git a/tests/migration-test.c b/tests/migration-test.c
> index 1c9f2c4e6a..fc221f172a 100644
> --- a/tests/migration-test.c
> +++ b/tests/migration-test.c
> @@ -460,6 +460,43 @@ static void migrate_set_parameter_int(QTestState *who, const char *parameter,
>      migrate_check_parameter_int(who, parameter, value);
>  }
>  
> +static char *migrate_get_parameter_str(QTestState *who,
> +                                       const char *parameter)
> +{
> +    QDict *rsp;
> +    char *result;
> +
> +    rsp = wait_command(who, "{ 'execute': 'query-migrate-parameters' }");
> +    result = g_strdup(qdict_get_str(rsp, parameter));
> +    qobject_unref(rsp);
> +    return result;
> +}
> +
> +static void migrate_check_parameter_str(QTestState *who, const char *parameter,
> +                                        const char *value)
> +{
> +    char *result;
> +
> +    result = migrate_get_parameter_str(who, parameter);
> +    g_assert_cmpstr(result, ==, value);
> +    g_free(result);
> +}
> +
> +__attribute__((unused))
> +static void migrate_set_parameter_str(QTestState *who, const char *parameter,
> +                                      const char *value)
> +{
> +    QDict *rsp;
> +
> +    rsp = qtest_qmp(who,
> +                    "{ 'execute': 'migrate-set-parameters',"
> +                    "'arguments': { %s: %s } }",
> +                    parameter, value);
> +    g_assert(qdict_haskey(rsp, "return"));
> +    qobject_unref(rsp);
> +    migrate_check_parameter_str(who, parameter, value);
> +}
> +
>  static void migrate_pause(QTestState *who)
>  {
>      QDict *rsp;
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Daniel P. Berrangé]

On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> We can scale much better with 16, so we can scale to higher numbers.

What was the test scenario showing such scaling ?

In the real world I'm sceptical that virt hosts will have
16 otherwise idle CPU cores available that are permissible
to use for migration, or indeed whether they'll have network
bandwidth available to allow 16 cores to saturate the link.


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 :|

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Dr. David Alan Gilbert]

* Daniel P. Berrangé (berrange@redhat.com) wrote:
> On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> > We can scale much better with 16, so we can scale to higher numbers.
> 
> What was the test scenario showing such scaling ?
> 
> In the real world I'm sceptical that virt hosts will have
> 16 otherwise idle CPU cores available that are permissible
> to use for migration, or indeed whether they'll have network
> bandwidth available to allow 16 cores to saturate the link.

With TLS or compression, the network bandwidth could easily be there.

Dave

> 
> 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 :|
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Daniel P. Berrangé]

On Fri, Jan 03, 2020 at 05:01:14PM +0000, Dr. David Alan Gilbert wrote:
> * Daniel P. Berrangé (berrange@redhat.com) wrote:
> > On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> > > We can scale much better with 16, so we can scale to higher numbers.
> > 
> > What was the test scenario showing such scaling ?
> > 
> > In the real world I'm sceptical that virt hosts will have
> > 16 otherwise idle CPU cores available that are permissible
> > to use for migration, or indeed whether they'll have network
> > bandwidth available to allow 16 cores to saturate the link.
> 
> With TLS or compression, the network bandwidth could easily be there.

Yes, but this constant is setting a default that applies regardless of
whether TLS / compression is enabled and/or whether CPU cores are idle.
IOW, there can be cases where using 16 threads will be a perf win, I'm
just questioning the suitability as a global default out of the box.

I feel like what's really lacking with migration is documentation
around the usefulness of the very many parameters, and the various
interesting combinations & tradeoffs around enabling them. So instead
of changing the default threads, can we focusing on improving
documentation so that mgmt apps have good information on which to
make the decision about whether & when to use 2 or 16 or $NNN migration
threads.

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 :|

Re: [PATCH v2 05/10] migration: Make multifd_load_setup() get an Error parameter

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> We need to change the full chain to pass the Error parameter.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/migration.c | 10 +++++-----
>  migration/migration.h |  2 +-
>  migration/ram.c       |  2 +-
>  migration/ram.h       |  2 +-
>  migration/rdma.c      |  2 +-
>  5 files changed, 9 insertions(+), 9 deletions(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index 5a56bd0c91..cf6cec5fb6 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -518,11 +518,11 @@ fail:
>      exit(EXIT_FAILURE);
>  }
>  
> -static void migration_incoming_setup(QEMUFile *f)
> +static void migration_incoming_setup(QEMUFile *f, Error **errp)
>  {
>      MigrationIncomingState *mis = migration_incoming_get_current();
>  
> -    if (multifd_load_setup() != 0) {
> +    if (multifd_load_setup(errp) != 0) {
>          /* We haven't been able to create multifd threads
>             nothing better to do */

But if you're taking an errp and the load fails, don't you want to
report the error before you exit? (with an error_get_pretty or
something?)

>          exit(EXIT_FAILURE);
> @@ -572,13 +572,13 @@ static bool postcopy_try_recover(QEMUFile *f)
>      return false;
>  }
>  
> -void migration_fd_process_incoming(QEMUFile *f)
> +void migration_fd_process_incoming(QEMUFile *f, Error **errp)
>  {
>      if (postcopy_try_recover(f)) {
>          return;
>      }
>  
> -    migration_incoming_setup(f);
> +    migration_incoming_setup(f, errp);
>      migration_incoming_process();

and if you're making incoming_setup able to fail, don't you need
to.... hmm, skip the incoming_process?

>  }
>  
> @@ -596,7 +596,7 @@ void migration_ioc_process_incoming(QIOChannel *ioc, Error **errp)
>              return;
>          }
>  
> -        migration_incoming_setup(f);
> +        migration_incoming_setup(f, errp);

Don't you need to make that use a local_err and propagate, like in the
other half of the if/else?

>          /*
>           * Common migration only needs one channel, so we can start
> diff --git a/migration/migration.h b/migration/migration.h
> index 79b3dda146..545f283ae7 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -265,7 +265,7 @@ struct MigrationState
>  
>  void migrate_set_state(int *state, int old_state, int new_state);
>  
> -void migration_fd_process_incoming(QEMUFile *f);
> +void migration_fd_process_incoming(QEMUFile *f, Error **errp);
>  void migration_ioc_process_incoming(QIOChannel *ioc, Error **errp);
>  void migration_incoming_process(void);
>  
> diff --git a/migration/ram.c b/migration/ram.c
> index 1f364cc23d..fcf50e648a 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -1421,7 +1421,7 @@ static void *multifd_recv_thread(void *opaque)
>      return NULL;
>  }
>  
> -int multifd_load_setup(void)
> +int multifd_load_setup(Error **errp)
>  {
>      int thread_count;
>      uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> diff --git a/migration/ram.h b/migration/ram.h
> index da22a417ea..42be471d52 100644
> --- a/migration/ram.h
> +++ b/migration/ram.h
> @@ -43,7 +43,7 @@ uint64_t ram_bytes_total(void);
>  
>  int multifd_save_setup(Error **errp);
>  void multifd_save_cleanup(void);
> -int multifd_load_setup(void);
> +int multifd_load_setup(Error **errp);
>  int multifd_load_cleanup(Error **errp);
>  bool multifd_recv_all_channels_created(void);
>  bool multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
> diff --git a/migration/rdma.c b/migration/rdma.c
> index e241dcb992..2379b8345b 100644
> --- a/migration/rdma.c
> +++ b/migration/rdma.c
> @@ -4004,7 +4004,7 @@ static void rdma_accept_incoming_migration(void *opaque)
>      }
>  
>      rdma->migration_started_on_destination = 1;
> -    migration_fd_process_incoming(f);
> +    migration_fd_process_incoming(f, errp);

Heck, the errp handling in rdma_accept_incoming_migration is very very
broken; I don't see how the errp ever gets reported or freed.
(But that's an existing problem)

>  }
>  
>  void rdma_start_incoming_migration(const char *host_port, Error **errp)
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Dr. David Alan Gilbert]

* Daniel P. Berrangé (berrange@redhat.com) wrote:
> On Fri, Jan 03, 2020 at 05:01:14PM +0000, Dr. David Alan Gilbert wrote:
> > * Daniel P. Berrangé (berrange@redhat.com) wrote:
> > > On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> > > > We can scale much better with 16, so we can scale to higher numbers.
> > > 
> > > What was the test scenario showing such scaling ?
> > > 
> > > In the real world I'm sceptical that virt hosts will have
> > > 16 otherwise idle CPU cores available that are permissible
> > > to use for migration, or indeed whether they'll have network
> > > bandwidth available to allow 16 cores to saturate the link.
> > 
> > With TLS or compression, the network bandwidth could easily be there.
> 
> Yes, but this constant is setting a default that applies regardless of
> whether TLS / compression is enabled and/or whether CPU cores are idle.
> IOW, there can be cases where using 16 threads will be a perf win, I'm
> just questioning the suitability as a global default out of the box.
> 
> I feel like what's really lacking with migration is documentation
> around the usefulness of the very many parameters, and the various
> interesting combinations & tradeoffs around enabling them. So instead
> of changing the default threads, can we focusing on improving
> documentation so that mgmt apps have good information on which to
> make the decision about whether & when to use 2 or 16 or $NNN migration
> threads.

Yes, although the short answer is; increase it if you find your
migration threads are saturated, either due to a very fast network
connection, or a CPU heavy setting (such as TLS or compression).
The answer to that might also vary if you have compression/encryption
offload engines (which I'd like to try).  Given that this series is for
compression, I guess that's the use Juan is using here.

On a 100Gbps NIC (which are readily available these days), I managed to
squeeze 70Gbps out of an earlier multifd version with 8 channels, which
beat the RDMA code in throughput (albeit eating CPU).

Dave

> 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 :|
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Daniel P. Berrangé]

On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> We can scale much better with 16, so we can scale to higher numbers.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/migration.c | 2 +-
>  1 file changed, 1 insertion(+), 1 deletion(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index 354ad072fa..e7f707e033 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -86,7 +86,7 @@
>  
>  /* The delay time (in ms) between two COLO checkpoints */
>  #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY (200 * 100)
> -#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 2
> +#define DEFAULT_MIGRATE_MULTIFD_CHANNELS 16

I think this can break compatibility for migration between new and
old QEMU. Consider a deployment has enabled multifd, but not given
an explicit number of channels, and now try to start the migration.

In the method multifd_recv_initial_packet() there is a check

    if (msg.id > migrate_multifd_channels()) {
        error_setg(errp, "multifd: received channel version %d "
                   "expected %d", msg.version, MULTIFD_VERSION);
        return -1;
    }

which will fail if migrating from new QEMU to old QEMU because
migrate_multifd_channels() will be 2 for old QEMU and new
QEMU will be trying to open 16 channels.

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 :|

Re: [PATCH v2 06/10] migration: Add multifd-compress parameter

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> ---
> Rename it to NONE
> Fix typos (dave)
> We don't need to chek values returned by visit_type_MultifdCompress (markus)
> Fix yet more typos (wei)
> ---
>  hw/core/qdev-properties.c    | 13 +++++++++++++
>  include/hw/qdev-properties.h |  3 +++
>  migration/migration.c        | 13 +++++++++++++
>  monitor/hmp-cmds.c           | 13 +++++++++++++
>  qapi/migration.json          | 30 +++++++++++++++++++++++++++---
>  tests/migration-test.c       | 13 ++++++++++---
>  6 files changed, 79 insertions(+), 6 deletions(-)
> 
> diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
> index ac28890e5a..644705235e 100644
> --- a/hw/core/qdev-properties.c
> +++ b/hw/core/qdev-properties.c
> @@ -8,6 +8,7 @@
>  #include "qapi/qmp/qerror.h"
>  #include "qemu/ctype.h"
>  #include "qemu/error-report.h"
> +#include "qapi/qapi-types-migration.h"
>  #include "hw/block/block.h"
>  #include "net/hub.h"
>  #include "qapi/visitor.h"
> @@ -648,6 +649,18 @@ const PropertyInfo qdev_prop_fdc_drive_type = {
>      .set_default_value = set_default_value_enum,
>  };
>  
> +/* --- MultifdCompress --- */
> +
> +const PropertyInfo qdev_prop_multifd_compress = {
> +    .name = "MultifdCompress",
> +    .description = "multifd_compress values, "
> +                   "none",
> +    .enum_table = &MultifdCompress_lookup,
> +    .get = get_enum,
> +    .set = set_enum,
> +    .set_default_value = set_default_value_enum,
> +};
> +
>  /* --- pci address --- */
>  
>  /*
> diff --git a/include/hw/qdev-properties.h b/include/hw/qdev-properties.h
> index c6a8cb5516..07d7bba682 100644
> --- a/include/hw/qdev-properties.h
> +++ b/include/hw/qdev-properties.h
> @@ -21,6 +21,7 @@ extern const PropertyInfo qdev_prop_tpm;
>  extern const PropertyInfo qdev_prop_ptr;
>  extern const PropertyInfo qdev_prop_macaddr;
>  extern const PropertyInfo qdev_prop_on_off_auto;
> +extern const PropertyInfo qdev_prop_multifd_compress;
>  extern const PropertyInfo qdev_prop_losttickpolicy;
>  extern const PropertyInfo qdev_prop_blockdev_on_error;
>  extern const PropertyInfo qdev_prop_bios_chs_trans;
> @@ -204,6 +205,8 @@ extern const PropertyInfo qdev_prop_pcie_link_width;
>      DEFINE_PROP(_n, _s, _f, qdev_prop_macaddr, MACAddr)
>  #define DEFINE_PROP_ON_OFF_AUTO(_n, _s, _f, _d) \
>      DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_on_off_auto, OnOffAuto)
> +#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
> +    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
>  #define DEFINE_PROP_LOSTTICKPOLICY(_n, _s, _f, _d) \
>      DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_losttickpolicy, \
>                          LostTickPolicy)
> diff --git a/migration/migration.c b/migration/migration.c
> index cf6cec5fb6..93c6ed10a6 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -87,6 +87,7 @@
>  /* The delay time (in ms) between two COLO checkpoints */
>  #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY (200 * 100)
>  #define DEFAULT_MIGRATE_MULTIFD_CHANNELS 16
> +#define DEFAULT_MIGRATE_MULTIFD_COMPRESS MULTIFD_COMPRESS_NONE
>  
>  /* Background transfer rate for postcopy, 0 means unlimited, note
>   * that page requests can still exceed this limit.
> @@ -774,6 +775,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
>      params->block_incremental = s->parameters.block_incremental;
>      params->has_multifd_channels = true;
>      params->multifd_channels = s->parameters.multifd_channels;
> +    params->has_multifd_compress = true;
> +    params->multifd_compress = s->parameters.multifd_compress;
>      params->has_xbzrle_cache_size = true;
>      params->xbzrle_cache_size = s->parameters.xbzrle_cache_size;
>      params->has_max_postcopy_bandwidth = true;
> @@ -1281,6 +1284,9 @@ static void migrate_params_test_apply(MigrateSetParameters *params,
>      if (params->has_multifd_channels) {
>          dest->multifd_channels = params->multifd_channels;
>      }
> +    if (params->has_multifd_compress) {
> +        dest->multifd_compress = params->multifd_compress;
> +    }
>      if (params->has_xbzrle_cache_size) {
>          dest->xbzrle_cache_size = params->xbzrle_cache_size;
>      }
> @@ -1377,6 +1383,9 @@ static void migrate_params_apply(MigrateSetParameters *params, Error **errp)
>      if (params->has_multifd_channels) {
>          s->parameters.multifd_channels = params->multifd_channels;
>      }
> +    if (params->has_multifd_compress) {
> +        s->parameters.multifd_compress = params->multifd_compress;
> +    }
>      if (params->has_xbzrle_cache_size) {
>          s->parameters.xbzrle_cache_size = params->xbzrle_cache_size;
>          xbzrle_cache_resize(params->xbzrle_cache_size, errp);
> @@ -3474,6 +3483,9 @@ static Property migration_properties[] = {
>      DEFINE_PROP_UINT8("multifd-channels", MigrationState,
>                        parameters.multifd_channels,
>                        DEFAULT_MIGRATE_MULTIFD_CHANNELS),
> +    DEFINE_PROP_MULTIFD_COMPRESS("multifd-compress", MigrationState,
> +                      parameters.multifd_compress,
> +                      DEFAULT_MIGRATE_MULTIFD_COMPRESS),
>      DEFINE_PROP_SIZE("xbzrle-cache-size", MigrationState,
>                        parameters.xbzrle_cache_size,
>                        DEFAULT_MIGRATE_XBZRLE_CACHE_SIZE),
> @@ -3564,6 +3576,7 @@ static void migration_instance_init(Object *obj)
>      params->has_x_checkpoint_delay = true;
>      params->has_block_incremental = true;
>      params->has_multifd_channels = true;
> +    params->has_multifd_compress = true;
>      params->has_xbzrle_cache_size = true;
>      params->has_max_postcopy_bandwidth = true;
>      params->has_max_cpu_throttle = true;
> diff --git a/monitor/hmp-cmds.c b/monitor/hmp-cmds.c
> index b2551c16d1..caf06b0668 100644
> --- a/monitor/hmp-cmds.c
> +++ b/monitor/hmp-cmds.c
> @@ -39,6 +39,7 @@
>  #include "qapi/qapi-commands-tpm.h"
>  #include "qapi/qapi-commands-ui.h"
>  #include "qapi/qapi-visit-net.h"
> +#include "qapi/qapi-visit-migration.h"
>  #include "qapi/qmp/qdict.h"
>  #include "qapi/qmp/qerror.h"
>  #include "qapi/string-input-visitor.h"
> @@ -448,6 +449,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
>          monitor_printf(mon, "%s: %u\n",
>              MigrationParameter_str(MIGRATION_PARAMETER_MULTIFD_CHANNELS),
>              params->multifd_channels);
> +        monitor_printf(mon, "%s: %s\n",
> +            MigrationParameter_str(MIGRATION_PARAMETER_MULTIFD_COMPRESS),
> +            MultifdCompress_str(params->multifd_compress));
>          monitor_printf(mon, "%s: %" PRIu64 "\n",
>              MigrationParameter_str(MIGRATION_PARAMETER_XBZRLE_CACHE_SIZE),
>              params->xbzrle_cache_size);
> @@ -1739,6 +1743,7 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
>      MigrateSetParameters *p = g_new0(MigrateSetParameters, 1);
>      uint64_t valuebw = 0;
>      uint64_t cache_size;
> +    MultifdCompress compress_type;
>      Error *err = NULL;
>      int val, ret;
>  
> @@ -1824,6 +1829,14 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
>          p->has_multifd_channels = true;
>          visit_type_int(v, param, &p->multifd_channels, &err);
>          break;
> +    case MIGRATION_PARAMETER_MULTIFD_COMPRESS:
> +        p->has_multifd_compress = true;
> +        visit_type_MultifdCompress(v, param, &compress_type, &err);
> +        if (err) {
> +            break;
> +        }
> +        p->multifd_compress = compress_type;
> +        break;
>      case MIGRATION_PARAMETER_XBZRLE_CACHE_SIZE:
>          p->has_xbzrle_cache_size = true;
>          visit_type_size(v, param, &cache_size, &err);
> diff --git a/qapi/migration.json b/qapi/migration.json
> index b7348d0c8b..430a39382e 100644
> --- a/qapi/migration.json
> +++ b/qapi/migration.json
> @@ -488,6 +488,19 @@
>  ##
>  { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
>  
> +##
> +# @MultifdCompress:
> +#
> +# An enumeration of multifd compression.
> +#
> +# @none: no compression.
> +#
> +# Since: 4.1
> +#
> +##
> +{ 'enum': 'MultifdCompress',
> +  'data': [ 'none' ] }
> +
>  ##
>  # @MigrationParameter:
>  #
> @@ -586,6 +599,9 @@
>  # @max-cpu-throttle: maximum cpu throttle percentage.
>  #                    Defaults to 99. (Since 3.1)
>  #
> +# @multifd-compress: Which compression method to use.
> +#                    Defaults to none. (Since 4.1)
> +#

5.0 I guess!

>  # Since: 2.4
>  ##
>  { 'enum': 'MigrationParameter',
> @@ -598,7 +614,7 @@
>             'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
>             'multifd-channels',
>             'xbzrle-cache-size', 'max-postcopy-bandwidth',
> -           'max-cpu-throttle' ] }
> +           'max-cpu-throttle', 'multifd-compress' ] }
>  
>  ##
>  # @MigrateSetParameters:
> @@ -688,6 +704,9 @@
>  # @max-cpu-throttle: maximum cpu throttle percentage.
>  #                    The default value is 99. (Since 3.1)
>  #
> +# @multifd-compress: Which compression method to use.
> +#                    Defaults to none. (Since 4.1)
> +#
>  # Since: 2.4
>  ##
>  # TODO either fuse back into MigrationParameters, or make
> @@ -713,7 +732,8 @@
>              '*multifd-channels': 'int',
>              '*xbzrle-cache-size': 'size',
>              '*max-postcopy-bandwidth': 'size',
> -	    '*max-cpu-throttle': 'int' } }
> +            '*max-cpu-throttle': 'int',
> +            '*multifd-compress': 'MultifdCompress' } }
>  
>  ##
>  # @migrate-set-parameters:
> @@ -823,6 +843,9 @@
>  #                    Defaults to 99.
>  #                     (Since 3.1)
>  #
> +# @multifd-compress: Which compression method to use.
> +#                    Defaults to none. (Since 4.1)
> +#
>  # Since: 2.4
>  ##
>  { 'struct': 'MigrationParameters',
> @@ -846,7 +869,8 @@
>              '*multifd-channels': 'uint8',
>              '*xbzrle-cache-size': 'size',
>  	    '*max-postcopy-bandwidth': 'size',
> -            '*max-cpu-throttle':'uint8'} }
> +            '*max-cpu-throttle': 'uint8',
> +            '*multifd-compress': 'MultifdCompress' } }
>  
>  ##
>  # @query-migrate-parameters:
> diff --git a/tests/migration-test.c b/tests/migration-test.c
> index fc221f172a..e5b6e54cfa 100644
> --- a/tests/migration-test.c
> +++ b/tests/migration-test.c
> @@ -482,7 +482,6 @@ static void migrate_check_parameter_str(QTestState *who, const char *parameter,
>      g_free(result);
>  }
>  
> -__attribute__((unused))
>  static void migrate_set_parameter_str(QTestState *who, const char *parameter,
>                                        const char *value)
>  {
> @@ -1398,7 +1397,7 @@ static void test_migrate_auto_converge(void)
>      test_migrate_end(from, to, true);
>  }
>  
> -static void test_multifd_tcp(void)
> +static void test_multifd_tcp(const char *method)
>  {
>      MigrateStart *args = migrate_start_new();
>      QTestState *from, *to;
> @@ -1422,6 +1421,9 @@ static void test_multifd_tcp(void)
>      migrate_set_parameter_int(from, "multifd-channels", 16);
>      migrate_set_parameter_int(to, "multifd-channels", 16);
>  
> +    migrate_set_parameter_str(from, "multifd-compress", method);
> +    migrate_set_parameter_str(to, "multifd-compress", method);
> +
>      migrate_set_capability(from, "multifd", "true");
>      migrate_set_capability(to, "multifd", "true");
>  
> @@ -1453,6 +1455,11 @@ static void test_multifd_tcp(void)
>      free(uri);
>  }
>  
> +static void test_multifd_tcp_none(void)
> +{
> +    test_multifd_tcp("none");
> +}
> +
>  int main(int argc, char **argv)
>  {
>      char template[] = "/tmp/migration-test-XXXXXX";
> @@ -1517,7 +1524,7 @@ int main(int argc, char **argv)
>                     test_validate_uuid_dst_not_set);
>  
>      qtest_add_func("/migration/auto_converge", test_migrate_auto_converge);
> -    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
> +    qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
>  
>      ret = g_test_run();

Except for the Since 4.1's:

Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

>  
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 07/10] migration: Make no compression operations into its own structure

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> It will be used later.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> ---
> Move setup of ->ops helper to proper place (wei)
> Rename s/none/nocomp/ (dave)
> Introduce MULTIFD_FLAG_NOCOMP
> ---
>  migration/migration.c |   9 ++
>  migration/migration.h |   1 +
>  migration/ram.c       | 194 ++++++++++++++++++++++++++++++++++++++++--
>  3 files changed, 196 insertions(+), 8 deletions(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index 93c6ed10a6..56203eb536 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -2213,6 +2213,15 @@ int migrate_multifd_channels(void)
>      return s->parameters.multifd_channels;
>  }
>  
> +int migrate_multifd_method(void)
> +{
> +    MigrationState *s;
> +
> +    s = migrate_get_current();
> +
> +    return s->parameters.multifd_compress;
> +}
> +
>  int migrate_use_xbzrle(void)
>  {
>      MigrationState *s;
> diff --git a/migration/migration.h b/migration/migration.h
> index 545f283ae7..d3ea45e25a 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -299,6 +299,7 @@ bool migrate_auto_converge(void);
>  bool migrate_use_multifd(void);
>  bool migrate_pause_before_switchover(void);
>  int migrate_multifd_channels(void);
> +int migrate_multifd_method(void);
>  
>  int migrate_use_xbzrle(void);
>  int64_t migrate_xbzrle_cache_size(void);
> diff --git a/migration/ram.c b/migration/ram.c
> index fcf50e648a..10661e03ae 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -44,6 +44,7 @@
>  #include "page_cache.h"
>  #include "qemu/error-report.h"
>  #include "qapi/error.h"
> +#include "qapi/qapi-types-migration.h"
>  #include "qapi/qapi-events-migration.h"
>  #include "qapi/qmp/qerror.h"
>  #include "trace.h"
> @@ -581,6 +582,7 @@ exit:
>  #define MULTIFD_VERSION 1
>  
>  #define MULTIFD_FLAG_SYNC (1 << 0)
> +#define MULTIFD_FLAG_NOCOMP (1 << 1)
>  
>  /* This value needs to be a multiple of qemu_target_page_size() */
>  #define MULTIFD_PACKET_SIZE (512 * 1024)
> @@ -662,6 +664,8 @@ typedef struct {
>      uint64_t num_pages;
>      /* syncs main thread and channels */
>      QemuSemaphore sem_sync;
> +    /* used for compression methods */
> +    void *data;
>  }  MultiFDSendParams;
>  
>  typedef struct {
> @@ -699,8 +703,153 @@ typedef struct {
>      uint64_t num_pages;
>      /* syncs main thread and channels */
>      QemuSemaphore sem_sync;
> +    /* used for de-compression methods */
> +    void *data;
>  } MultiFDRecvParams;
>  
> +typedef struct {
> +    /* Setup for sending side */
> +    int (*send_setup)(MultiFDSendParams *p, Error **errp);
> +    /* Cleanup for sending side */
> +    void (*send_cleanup)(MultiFDSendParams *p, Error **errp);
> +    /* Prepare the send packet */
> +    int (*send_prepare)(MultiFDSendParams *p, uint32_t used, Error **errp);
> +    /* Write the send packet */
> +    int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **errp);
> +    /* Setup for receiving side */
> +    int (*recv_setup)(MultiFDRecvParams *p, Error **errp);
> +    /* Cleanup for receiving side */
> +    void (*recv_cleanup)(MultiFDRecvParams *p);
> +    /* Read all pages */
> +    int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **errp);
> +} MultiFDMethods;
> +
> +/* Multifd without compression */
> +
> +/**
> + * nocomp_send_setup: setup send side
> + *
> + * For no compression this function does nothing.
> + *
> + * Returns 0 for success or -1 for error
> + *
> + * @p: Params for the channel that we are using
> + * @errp: pointer to an error
> + */
> +static int nocomp_send_setup(MultiFDSendParams *p, Error **errp)
> +{
> +    return 0;
> +}
> +
> +/**
> + * nocomp_send_cleanup: cleanup send side
> + *
> + * For no compression this function does nothing.
> + *
> + * @p: Params for the channel that we are using
> + */
> +static void nocomp_send_cleanup(MultiFDSendParams *p, Error **errp)
> +{
> +    return;
> +}
> +
> +/**
> + * nocomp_send_prepare: prepare date to be able to send
> + *
> + * For no compression we just have to calculate the size of the
> + * packet.
> + *
> + * Returns 0 for success or -1 for error
> + *
> + * @p: Params for the channel that we are using
> + * @used: number of pages used
> + * @errp: pointer to an error
> + */
> +static int nocomp_send_prepare(MultiFDSendParams *p, uint32_t used,
> +                               Error **errp)
> +{
> +    p->next_packet_size = used * qemu_target_page_size();
> +    p->flags |= MULTIFD_FLAG_NOCOMP;
> +    return 0;
> +}
> +
> +/**
> + * nocomp_send_write: do the actual write of the data
> + *
> + * For no compression we just have to write the data.
> + *
> + * Returns 0 for success or -1 for error
> + *
> + * @p: Params for the channel that we are using
> + * @used: number of pages used
> + * @errp: pointer to an error
> + */
> +static int nocomp_send_write(MultiFDSendParams *p, uint32_t used, Error **errp)
> +{
> +    return qio_channel_writev_all(p->c, p->pages->iov, used, errp);
> +}
> +
> +/**
> + * nocomp_recv_setup: setup receive side
> + *
> + * For no compression this function does nothing.
> + *
> + * Returns 0 for success or -1 for error
> + *
> + * @p: Params for the channel that we are using
> + * @errp: pointer to an error
> + */
> +static int nocomp_recv_setup(MultiFDRecvParams *p, Error **errp)
> +{
> +    return 0;
> +}
> +
> +/**
> + * nocomp_recv_cleanup: setup receive side
> + *
> + * For no compression this function does nothing.
> + *
> + * @p: Params for the channel that we are using
> + */
> +static void nocomp_recv_cleanup(MultiFDRecvParams *p)
> +{
> +}
> +
> +/**
> + * nocomp_recv_pages: read the data from the channel into actual pages
> + *
> + * For no compression we just need to read things into the correct place.
> + *
> + * Returns 0 for success or -1 for error
> + *
> + * @p: Params for the channel that we are using
> + * @used: number of pages used
> + * @errp: pointer to an error
> + */
> +static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
> +{
> +    if (p->flags != MULTIFD_FLAG_NOCOMP) {
> +        error_setg(errp, "multifd %d: flags received %x flags expected %x",
> +                   p->id, MULTIFD_FLAG_NOCOMP, p->flags);

That looks the wrong way around to me - shouldn't that be
   p->flags, MULTIFD_FLAG_NOCOMP
?

> +        return -1;
> +    }
> +    return qio_channel_readv_all(p->c, p->pages->iov, used, errp);
> +}
> +
> +static MultiFDMethods multifd_nocomp_ops = {
> +    .send_setup = nocomp_send_setup,
> +    .send_cleanup = nocomp_send_cleanup,
> +    .send_prepare = nocomp_send_prepare,
> +    .send_write = nocomp_send_write,
> +    .recv_setup = nocomp_recv_setup,
> +    .recv_cleanup = nocomp_recv_cleanup,
> +    .recv_pages = nocomp_recv_pages
> +};
> +
> +static MultiFDMethods *multifd_ops[MULTIFD_COMPRESS__MAX] = {
> +    [MULTIFD_COMPRESS_NONE] = &multifd_nocomp_ops,
> +};
> +
>  static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
>  {
>      MultiFDInit_t msg;
> @@ -898,6 +1047,8 @@ struct {
>      uint64_t packet_num;
>      /* send channels ready */
>      QemuSemaphore channels_ready;
> +    /* multifd ops */
> +    MultiFDMethods *ops;
>  } *multifd_send_state;
>  
>  /*
> @@ -1027,6 +1178,7 @@ void multifd_save_cleanup(void)
>      multifd_send_terminate_threads(NULL);
>      for (i = 0; i < migrate_multifd_channels(); i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
> +        Error *local_err = NULL;
>  
>          if (p->running) {
>              qemu_thread_join(&p->thread);
> @@ -1043,6 +1195,10 @@ void multifd_save_cleanup(void)
>          p->packet_len = 0;
>          g_free(p->packet);
>          p->packet = NULL;
> +        multifd_send_state->ops->send_cleanup(p, &local_err);
> +        if (local_err) {
> +            migrate_set_error(migrate_get_current(), local_err);
> +        }
>      }
>      qemu_sem_destroy(&multifd_send_state->channels_ready);
>      g_free(multifd_send_state->params);
> @@ -1123,7 +1279,14 @@ static void *multifd_send_thread(void *opaque)
>              uint64_t packet_num = p->packet_num;
>              flags = p->flags;
>  
> -            p->next_packet_size = used * qemu_target_page_size();
> +            if (used) {
> +                ret = multifd_send_state->ops->send_prepare(p, used,
> +                                                            &local_err);
> +                if (ret != 0) {
> +                    qemu_mutex_unlock(&p->mutex);
> +                    break;
> +                }
> +            }
>              multifd_send_fill_packet(p);
>              p->flags = 0;
>              p->num_packets++;
> @@ -1140,8 +1303,7 @@ static void *multifd_send_thread(void *opaque)
>              }
>  
>              if (used) {
> -                ret = qio_channel_writev_all(p->c, p->pages->iov,
> -                                             used, &local_err);
> +                ret = multifd_send_state->ops->send_write(p, used, &local_err);
>                  if (ret != 0) {
>                      break;
>                  }
> @@ -1212,6 +1374,7 @@ int multifd_save_setup(Error **errp)
>  {
>      int thread_count;
>      uint32_t page_count = MULTIFD_PACKET_SIZE / qemu_target_page_size();
> +    int ret = 0;
>      uint8_t i;
>  
>      if (!migrate_use_multifd()) {
> @@ -1222,9 +1385,11 @@ int multifd_save_setup(Error **errp)
>      multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
>      multifd_send_state->pages = multifd_pages_init(page_count);
>      qemu_sem_init(&multifd_send_state->channels_ready, 0);
> +    multifd_send_state->ops = multifd_ops[migrate_multifd_method()];
>  
>      for (i = 0; i < thread_count; i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
> +        int res;
>  
>          qemu_mutex_init(&p->mutex);
>          qemu_sem_init(&p->sem, 0);
> @@ -1240,8 +1405,12 @@ int multifd_save_setup(Error **errp)
>          p->packet->version = cpu_to_be32(MULTIFD_VERSION);
>          p->name = g_strdup_printf("multifdsend_%d", i);
>          socket_send_channel_create(multifd_new_send_channel_async, p);
> +        res = multifd_send_state->ops->send_setup(p, errp);
> +        if (ret == 0) {
> +            ret = res;

How do you handle the errp's here - I think that if is so that you
get the 'ret' from the first thread that fails; but I don't think you're
allowed to call that twice if the first one set it's errp.

> +        }
>      }
> -    return 0;
> +    return ret;
>  }
>  
>  struct {
> @@ -1252,6 +1421,8 @@ struct {
>      QemuSemaphore sem_sync;
>      /* global number of generated multifd packets */
>      uint64_t packet_num;
> +    /* multifd ops */
> +    MultiFDMethods *ops;
>  } *multifd_recv_state;
>  
>  static void multifd_recv_terminate_threads(Error *err)
> @@ -1287,7 +1458,6 @@ static void multifd_recv_terminate_threads(Error *err)
>  int multifd_load_cleanup(Error **errp)
>  {
>      int i;
> -    int ret = 0;
>  
>      if (!migrate_use_multifd()) {
>          return 0;
> @@ -1316,6 +1486,7 @@ int multifd_load_cleanup(Error **errp)
>          p->packet_len = 0;
>          g_free(p->packet);
>          p->packet = NULL;
> +        multifd_recv_state->ops->recv_cleanup(p);
>      }
>      qemu_sem_destroy(&multifd_recv_state->sem_sync);
>      g_free(multifd_recv_state->params);
> @@ -1323,7 +1494,7 @@ int multifd_load_cleanup(Error **errp)
>      g_free(multifd_recv_state);
>      multifd_recv_state = NULL;
>  
> -    return ret;
> +    return 0;
>  }
>  
>  static void multifd_recv_sync_main(void)
> @@ -1388,6 +1559,8 @@ static void *multifd_recv_thread(void *opaque)
>  
>          used = p->pages->used;
>          flags = p->flags;
> +        /* recv methods don't know how to handle the SYNC flag */
> +        p->flags &= ~MULTIFD_FLAG_SYNC;
>          trace_multifd_recv(p->id, p->packet_num, used, flags,
>                             p->next_packet_size);
>          p->num_packets++;
> @@ -1395,8 +1568,7 @@ static void *multifd_recv_thread(void *opaque)
>          qemu_mutex_unlock(&p->mutex);
>  
>          if (used) {
> -            ret = qio_channel_readv_all(p->c, p->pages->iov,
> -                                        used, &local_err);
> +            ret = multifd_recv_state->ops->recv_pages(p, used, &local_err);
>              if (ret != 0) {
>                  break;
>              }
> @@ -1435,9 +1607,11 @@ int multifd_load_setup(Error **errp)
>      multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
>      atomic_set(&multifd_recv_state->count, 0);
>      qemu_sem_init(&multifd_recv_state->sem_sync, 0);
> +    multifd_recv_state->ops = multifd_ops[migrate_multifd_method()];
>  
>      for (i = 0; i < thread_count; i++) {
>          MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +        int ret;
>  
>          qemu_mutex_init(&p->mutex);
>          qemu_sem_init(&p->sem_sync, 0);
> @@ -1448,6 +1622,10 @@ int multifd_load_setup(Error **errp)
>                        + sizeof(ram_addr_t) * page_count;
>          p->packet = g_malloc0(p->packet_len);
>          p->name = g_strdup_printf("multifdrecv_%d", i);
> +        ret = multifd_recv_state->ops->recv_setup(p, errp);
> +        if (ret != 0) {
> +            return ret;
> +        }

same question as the save case above

>      }
>      return 0;
>  }
> -- 
> 2.23.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Juan Quintela]

Daniel P. Berrangé <berrange@redhat.com> wrote:
> On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
>> We can scale much better with 16, so we can scale to higher numbers.
>
> What was the test scenario showing such scaling ?

On my test hardware, with 2 channels we can saturate around 8Gigabit max,
more than that, and the migration thread is not fast enough to fill the
network bandwidth.

With 8 that is enough to fill whatever we can find.
We used to have a bug where we were getting trouble with more channels
than cores.  That was the initial reason why the default was so low.

So, pros/cons are:
- have low value (2).  We are backwards compatible, but we are not using
  all  bandwith.  Notice that we will dectect the error before 5.0 is
  out and print a good error message.

- have high value (I tested 8 and 16).  Found no performance loss when
  moving to lower bandwidth limits, and clearly we were able to saturate
  the higher speeds (I tested on localhost, so I had big enough bandwidth)


> In the real world I'm sceptical that virt hosts will have
> 16 otherwise idle CPU cores available that are permissible
> to use for migration, or indeed whether they'll have network
> bandwidth available to allow 16 cores to saturate the link.

The problem here is that if you have such a host, and you want to have
high speed migration, you need to configure it.  My measumermets are
that high number of channels don't affect performance with low
bandwidth, but low number of channels affect performance with high
bandwidth speed.

So, if we want to have something that works "automatically" everywhere,
we need to put it to at least 8.  Or we can trust that management app
will do the right thing.

If you are using a low value of bandwidth, the only difference with 16
channels is that you are using a bit more memory (just the space for the
stacks) and that you are having less contention for the locks (but with
low bandwidth you are not having contention anyways).

So,  I think that the question is:
- What does libvirt prefferes
- What does ovirt/openstack preffer
- Do we really want that the user "have" to configure that value

I don't really care one way or another.

Thanks, Juan.

PD.  On next patch submission I will make it be 2 for old machine types,
     it is not difficult and makes the backward compatibility problem go
     away.

Re: [PATCH v2 04/10] migration: Make multifd_save_setup() get an Error parameter

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/migration.c | 2 +-
>>  migration/ram.c       | 2 +-
>>  migration/ram.h       | 2 +-
>>  3 files changed, 3 insertions(+), 3 deletions(-)
>> 
>> diff --git a/migration/migration.c b/migration/migration.c
>> index e7f707e033..5a56bd0c91 100644
>> --- a/migration/migration.c
>> +++ b/migration/migration.c
>> @@ -3400,7 +3400,7 @@ void migrate_fd_connect(MigrationState *s, Error *error_in)
>>          return;
>>      }
>>  
>> -    if (multifd_save_setup() != 0) {
>> +    if (multifd_save_setup(&error_in) != 0) {
>
> I'm not sure that's right.  I think the *error passed into
> migration_channel_connect, and then onto migrate_fd_connect is an
> indication that an error has happened, not a place you can put
> an error pointer.   Note how migration_channel_connect
> frees it after the migrate_fd_connect call, it doesn't report it.


changed this to:

    if (multifd_save_setup(&local_err) != 0) {
       error_report_err(local_err);

Thanks, Juan.

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Daniel P. Berrangé]

On Fri, Jan 03, 2020 at 07:25:08PM +0100, Juan Quintela wrote:
> Daniel P. Berrangé <berrange@redhat.com> wrote:
> > On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
> >> We can scale much better with 16, so we can scale to higher numbers.
> >
> > What was the test scenario showing such scaling ?
> 
> On my test hardware, with 2 channels we can saturate around 8Gigabit max,
> more than that, and the migration thread is not fast enough to fill the
> network bandwidth.
> 
> With 8 that is enough to fill whatever we can find.
> We used to have a bug where we were getting trouble with more channels
> than cores.  That was the initial reason why the default was so low.
> 
> So, pros/cons are:
> - have low value (2).  We are backwards compatible, but we are not using
>   all  bandwith.  Notice that we will dectect the error before 5.0 is
>   out and print a good error message.
> 
> - have high value (I tested 8 and 16).  Found no performance loss when
>   moving to lower bandwidth limits, and clearly we were able to saturate
>   the higher speeds (I tested on localhost, so I had big enough bandwidth)
> 
> 
> > In the real world I'm sceptical that virt hosts will have
> > 16 otherwise idle CPU cores available that are permissible
> > to use for migration, or indeed whether they'll have network
> > bandwidth available to allow 16 cores to saturate the link.
> 
> The problem here is that if you have such a host, and you want to have
> high speed migration, you need to configure it.  My measumermets are
> that high number of channels don't affect performance with low
> bandwidth, but low number of channels affect performance with high
> bandwidth speed.

I'm not concerned about impact on performance of migration on a
low bandwidth link, rather I'm concerned about impact on performance
of other guests on the host. It will cause migration to contend with
other guest's vCPUs and network traffic. 

> So, if we want to have something that works "automatically" everywhere,
> we need to put it to at least 8.  Or we can trust that management app
> will do the right thing.

Aren't we still setting the bandwidth limit to MB bandwidth out of the
box, so we already require mgmt app to change settings to use more
bandwidth ? 

> If you are using a low value of bandwidth, the only difference with 16
> channels is that you are using a bit more memory (just the space for the
> stacks) and that you are having less contention for the locks (but with
> low bandwidth you are not having contention anyways).
> 
> So,  I think that the question is:
> - What does libvirt prefferes

Libvirt doesn't really have an opinion in this case. I believe we'll
always set the number of channels on both src & dst, so we don't
see the defaults.

> - What does ovirt/openstack preffer

Libvirt should insulate them from any change in defaults in QEMU
in this case, but always explicitly setting channels on src & dst
to match.

> - Do we really want that the user "have" to configure that value

Right so this is the key quesiton - for a user not using libvirt
or a libvirt based mgmt app, what we do want out out of the box
migration to be tuned for ?

If we want to maximise migration performance, at cost of anything
else, then we can change the migration channels count, but probably
also ought to remove the 32MB bandwidth cap as no useful guest with
active apps will succeed migration with a 32MB cap.

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 :|

Re: [PATCH v2 05/10] migration: Make multifd_load_setup() get an Error parameter

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> We need to change the full chain to pass the Error parameter.
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/migration.c | 10 +++++-----
>>  migration/migration.h |  2 +-
>>  migration/ram.c       |  2 +-
>>  migration/ram.h       |  2 +-
>>  migration/rdma.c      |  2 +-
>>  5 files changed, 9 insertions(+), 9 deletions(-)
>> 
>> diff --git a/migration/migration.c b/migration/migration.c
>> index 5a56bd0c91..cf6cec5fb6 100644
>> --- a/migration/migration.c
>> +++ b/migration/migration.c
>> @@ -518,11 +518,11 @@ fail:
>>      exit(EXIT_FAILURE);
>>  }
>>  
>> -static void migration_incoming_setup(QEMUFile *f)
>> +static void migration_incoming_setup(QEMUFile *f, Error **errp)
>>  {
>>      MigrationIncomingState *mis = migration_incoming_get_current();
>>  
>> -    if (multifd_load_setup() != 0) {
>> +    if (multifd_load_setup(errp) != 0) {
>>          /* We haven't been able to create multifd threads
>>             nothing better to do */
>
> But if you're taking an errp and the load fails, don't you want to
> report the error before you exit? (with an error_get_pretty or
> something?)

error_report_err() that is.

>
>>          exit(EXIT_FAILURE);
>> @@ -572,13 +572,13 @@ static bool postcopy_try_recover(QEMUFile *f)
>>      return false;
>>  }
>>  
>> -void migration_fd_process_incoming(QEMUFile *f)
>> +void migration_fd_process_incoming(QEMUFile *f, Error **errp)
>>  {
>>      if (postcopy_try_recover(f)) {
>>          return;
>>      }
>>  
>> -    migration_incoming_setup(f);
>> +    migration_incoming_setup(f, errp);
>>      migration_incoming_process();
>
> and if you're making incoming_setup able to fail, don't you need
> to.... hmm, skip the incoming_process?

Changing it to test for errors, thanks.

>>  }
>>  
>> @@ -596,7 +596,7 @@ void migration_ioc_process_incoming(QIOChannel *ioc, Error **errp)
>>              return;
>>          }
>>  
>> -        migration_incoming_setup(f);
>> +        migration_incoming_setup(f, errp);
>
> Don't you need to make that use a local_err and propagate, like in the
> other half of the if/else?

Changed the whole business.  It appears that each time that I use an
Error ** I have to relearn how to use it.  Changed all places to use a
local error and propagate/error_report_err() as apropiate.

>>      rdma->migration_started_on_destination = 1;
>> -    migration_fd_process_incoming(f);
>> +    migration_fd_process_incoming(f, errp);
>
> Heck, the errp handling in rdma_accept_incoming_migration is very very
> broken; I don't see how the errp ever gets reported or freed.
> (But that's an existing problem)

Leaving this for next series.

Thanks, Juan.

Re: [PATCH v2 06/10] migration: Add multifd-compress parameter

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:

>> +#
>> +##
>> +{ 'enum': 'MultifdCompress',
>> +  'data': [ 'none' ] }
>> +
>>  ##
>>  # @MigrationParameter:
>>  #
>> @@ -586,6 +599,9 @@
>>  # @max-cpu-throttle: maximum cpu throttle percentage.
>>  #                    Defaults to 99. (Since 3.1)
>>  #
>> +# @multifd-compress: Which compression method to use.
>> +#                    Defaults to none. (Since 4.1)
>> +#
>
> 5.0 I guess!

changed.

>
> Except for the Since 4.1's:
>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>

Thanks, Juan.

Re: [PATCH v2 07/10] migration: Make no compression operations into its own structure

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> It will be used later.
>> 
>> +static int nocomp_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **errp)
>> +{
>> +    if (p->flags != MULTIFD_FLAG_NOCOMP) {
>> +        error_setg(errp, "multifd %d: flags received %x flags expected %x",
>> +                   p->id, MULTIFD_FLAG_NOCOMP, p->flags);
>
> That looks the wrong way around to me - shouldn't that be
>    p->flags, MULTIFD_FLAG_NOCOMP
> ?

Good catch.
>>  
>>          qemu_mutex_init(&p->mutex);
>>          qemu_sem_init(&p->sem, 0);
>> @@ -1240,8 +1405,12 @@ int multifd_save_setup(Error **errp)
>>          p->packet->version = cpu_to_be32(MULTIFD_VERSION);
>>          p->name = g_strdup_printf("multifdsend_%d", i);
>>          socket_send_channel_create(multifd_new_send_channel_async, p);
>> +        res = multifd_send_state->ops->send_setup(p, errp);
>> +        if (ret == 0) {
>> +            ret = res;
>
> How do you handle the errp's here - I think that if is so that you
> get the 'ret' from the first thread that fails; but I don't think you're
> allowed to call that twice if the first one set it's errp.

You are right.  I was doing the res/ret variable right, and failed with
the other.  Changed the code to two loops:
- Everything that can't fail, done in the 1st loop.
- Everything else on the second loop.  the time that we have one error,
  we just stop the loop.
>> @@ -1448,6 +1622,10 @@ int multifd_load_setup(Error **errp)
>>                        + sizeof(ram_addr_t) * page_count;
>>          p->packet = g_malloc0(p->packet_len);
>>          p->name = g_strdup_printf("multifdrecv_%d", i);
>> +        ret = multifd_recv_state->ops->recv_setup(p, errp);
>> +        if (ret != 0) {
>> +            return ret;
>> +        }
>
> same question as the save case above

Same solution.

Thanks, Juan.

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Juan Quintela]

Daniel P. Berrangé <berrange@redhat.com> wrote:
> On Fri, Jan 03, 2020 at 07:25:08PM +0100, Juan Quintela wrote:
>> Daniel P. Berrangé <berrange@redhat.com> wrote:
>> > On Wed, Dec 18, 2019 at 03:01:10AM +0100, Juan Quintela wrote:
>> >> We can scale much better with 16, so we can scale to higher numbers.
>> >
>> > What was the test scenario showing such scaling ?
>> 
>> On my test hardware, with 2 channels we can saturate around 8Gigabit max,
>> more than that, and the migration thread is not fast enough to fill the
>> network bandwidth.
>> 
>> With 8 that is enough to fill whatever we can find.
>> We used to have a bug where we were getting trouble with more channels
>> than cores.  That was the initial reason why the default was so low.
>> 
>> So, pros/cons are:
>> - have low value (2).  We are backwards compatible, but we are not using
>>   all  bandwith.  Notice that we will dectect the error before 5.0 is
>>   out and print a good error message.
>> 
>> - have high value (I tested 8 and 16).  Found no performance loss when
>>   moving to lower bandwidth limits, and clearly we were able to saturate
>>   the higher speeds (I tested on localhost, so I had big enough bandwidth)
>> 
>> 
>> > In the real world I'm sceptical that virt hosts will have
>> > 16 otherwise idle CPU cores available that are permissible
>> > to use for migration, or indeed whether they'll have network
>> > bandwidth available to allow 16 cores to saturate the link.
>> 
>> The problem here is that if you have such a host, and you want to have
>> high speed migration, you need to configure it.  My measumermets are
>> that high number of channels don't affect performance with low
>> bandwidth, but low number of channels affect performance with high
>> bandwidth speed.
>
> I'm not concerned about impact on performance of migration on a
> low bandwidth link, rather I'm concerned about impact on performance
> of other guests on the host. It will cause migration to contend with
> other guest's vCPUs and network traffic. 

Two things here:
- vcpus:  If you want migration to consume all the bandwidth, you are
  happy with it using more vcpus.
- bandwidth: It will only consume only the one that the guest has
  assigned, split (we hope evenly) between all the channels.

My main reason to have a higher number of channels is:
- test better the code with more than one channel
- work "magically" well in all scenarios.  With a low number of
  channels, we are not going to be able to saturate a big network pipe.


>
>> So, if we want to have something that works "automatically" everywhere,
>> we need to put it to at least 8.  Or we can trust that management app
>> will do the right thing.
>
> Aren't we still setting the bandwidth limit to MB bandwidth out of the
> box, so we already require mgmt app to change settings to use more
> bandwidth ? 

Yeap.  This is the default bandwidth.

#define MAX_THROTTLE  (32 << 20)


>> If you are using a low value of bandwidth, the only difference with 16
>> channels is that you are using a bit more memory (just the space for the
>> stacks) and that you are having less contention for the locks (but with
>> low bandwidth you are not having contention anyways).
>> 
>> So,  I think that the question is:

Note that my idea is to make multifd "default" in the near future (5.1
timeframe or so).

>> - What does libvirt prefferes
>
> Libvirt doesn't really have an opinion in this case. I believe we'll
> always set the number of channels on both src & dst, so we don't
> see the defaults.

What does libvirt does today for this value?

>> - What does ovirt/openstack preffer
>
> Libvirt should insulate them from any change in defaults in QEMU
> in this case, but always explicitly setting channels on src & dst
> to match.

I agree here, they should don't care by default.

>> - Do we really want that the user "have" to configure that value
>
> Right so this is the key quesiton - for a user not using libvirt
> or a libvirt based mgmt app, what we do want out out of the box
> migration to be tuned for ?

In my opinion, we should have something like:
- multifd: enabled by default
- max downtime: 300 ms (current) looks right to me
- max bandwidth: 32MB/s (current) seems a bit low. 100MB/s (i.e. almost
  full gigabit ethernet) seems reasonable to me.  Having a default for
  10Gigabit ethernet or similar seems too high.

> If we want to maximise migration performance, at cost of anything
> else, then we can change the migration channels count, but probably
> also ought to remove the 32MB bandwidth cap as no useful guest with
> active apps will succeed migration with a 32MB cap.

Will start another series with the current values to discuss all the
defaults, ok?

thanks for the comments, Juan.

Re: [PATCH v2 01/10] migration: Increase default number of multifd channels to 16

[Daniel P. Berrangé]

On Tue, Jan 07, 2020 at 02:32:24PM +0100, Juan Quintela wrote:
> Daniel P. Berrangé <berrange@redhat.com> wrote:
> >> - What does libvirt prefferes
> >
> > Libvirt doesn't really have an opinion in this case. I believe we'll
> > always set the number of channels on both src & dst, so we don't
> > see the defaults.
> 
> What does libvirt does today for this value?

Libvirt doesn't have any default value. By default migration uses a
single connection.  The mgmt app can request multiple connections
for the migration, which will cause libvirt to turn on multifd with
the request channel count.

Essentially we avoid the problem by not having a separate tunable
for "multifd enable" & "multifd channels" - we only have one tunable
"connection count".  Thus the idea of enabling multifd with a default
channel count doesn't exist.


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 :|