[Qemu-devel] [PATCH v3 0/8] WIP: Multifd compression support

[Qemu-devel] [PATCH v3 0/8] WIP: Multifd compression support

[Juan Quintela]

v3:
- improve the code
- address David and Markus comments
- make compression code into methods
  so we can add any other method ading just three functions

Please review, as far as I know everything is ok now.

Todo: Add zstd support

v2:
- improve the code left and right
- Split better the zlib code
- rename everything to v4.1
- Add tests for multifd-compress zlib
- Parameter is now an enum (soon will see sztd)

ToDo:
- Make operations for diferent methods:
  * multifd_prepare_send_none/zlib
  * multifd_send_none/zlib
  * multifd_recv_none/zlib
- Use the MULTIFD_FLAG_ZLIB (it is unused so far).

Please review and comment.

v1:

This series create compression code on top of multifd.  It is still
WIP, but it is already:
- faster that current compression code
- it does the minimum amount of copies possible
- we allow support for other compression codes
- it pass the multifd test sent in my previous series

Test for existing code didn't work because code is too slow, I need to
make downtime 10 times bigger to make it to converge on my test
machine.  This code works with same limits that multifd no-

ToDo:
- move printf's  to traces
- move code to a struct instead of if (zlib) inside the main threads.
- improve error handling.

Please, review and coment.

Juan Quintela (8):
  migration: fix multifd_recv event typo
  migration-test: rename parameter to parameter_int
  tests: Add migration multifd test
  migration-test: introduce functions to handle string parameters
  migration: Add multifd-compress parameter
  migration: Make none operations into its own structure
  multifd: Add zlib compression support
  multifd: rest of zlib compression

 hmp.c                        |  17 +++
 hw/core/qdev-properties.c    |  13 +++
 include/hw/qdev-properties.h |   1 +
 migration/migration.c        |  25 +++++
 migration/migration.h        |   1 +
 migration/ram.c              | 203 ++++++++++++++++++++++++++++++++++-
 migration/trace-events       |   2 +-
 qapi/migration.json          |  30 +++++-
 tests/migration-test.c       | 147 ++++++++++++++++++++-----
 9 files changed, 406 insertions(+), 33 deletions(-)

-- 
2.21.0

[Qemu-devel] [PATCH v3 1/8] migration: fix multifd_recv event typo

[Juan Quintela]

It uses num in multifd_send().  Make it coherent.

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

diff --git a/migration/trace-events b/migration/trace-events
index de2e136e57..cd50a1e659 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -80,7 +80,7 @@ get_queued_page_not_dirty(const char *block_name, uint64_t tmp_offset, unsigned
 migration_bitmap_sync_start(void) ""
 migration_bitmap_sync_end(uint64_t dirty_pages) "dirty_pages %" PRIu64
 migration_throttle(void) ""
-multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet number %" PRIu64 " pages %d flags 0x%x next packet size %d"
+multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet_num %" PRIu64 " pages %d flags 0x%x next packet size %d"
 multifd_recv_sync_main(long packet_num) "packet num %ld"
 multifd_recv_sync_main_signal(uint8_t id) "channel %d"
 multifd_recv_sync_main_wait(uint8_t id) "channel %d"
-- 
2.21.0

[Qemu-devel] [PATCH v3 2/8] migration-test: rename parameter to parameter_int

[Juan Quintela]

We would need _str ones on the next patch.

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

diff --git a/tests/migration-test.c b/tests/migration-test.c
index bd3f5c3125..0b25aa3d6c 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -392,7 +392,8 @@ static char *migrate_get_socket_address(QTestState *who, const char *parameter)
     return result;
 }
 
-static long long migrate_get_parameter(QTestState *who, const char *parameter)
+static long long migrate_get_parameter_int(QTestState *who,
+                                           const char *parameter)
 {
     QDict *rsp;
     long long result;
@@ -403,17 +404,17 @@ static long long migrate_get_parameter(QTestState *who, const char *parameter)
     return result;
 }
 
-static void migrate_check_parameter(QTestState *who, const char *parameter,
-                                    long long value)
+static void migrate_check_parameter_int(QTestState *who, const char *parameter,
+                                        long long value)
 {
     long long result;
 
-    result = migrate_get_parameter(who, parameter);
+    result = migrate_get_parameter_int(who, parameter);
     g_assert_cmpint(result, ==, value);
 }
 
-static void migrate_set_parameter(QTestState *who, const char *parameter,
-                                  long long value)
+static void migrate_set_parameter_int(QTestState *who, const char *parameter,
+                                      long long value)
 {
     QDict *rsp;
 
@@ -423,7 +424,7 @@ static void migrate_set_parameter(QTestState *who, const char *parameter,
                     parameter, value);
     g_assert(qdict_haskey(rsp, "return"));
     qobject_unref(rsp);
-    migrate_check_parameter(who, parameter, value);
+    migrate_check_parameter_int(who, parameter, value);
 }
 
 static void migrate_pause(QTestState *who)
@@ -672,7 +673,7 @@ static void deprecated_set_downtime(QTestState *who, const double value)
                     " 'arguments': { 'value': %f } }", value);
     g_assert(qdict_haskey(rsp, "return"));
     qobject_unref(rsp);
-    migrate_check_parameter(who, "downtime-limit", value * 1000);
+    migrate_check_parameter_int(who, "downtime-limit", value * 1000);
 }
 
 static void deprecated_set_speed(QTestState *who, long long value)
@@ -683,7 +684,7 @@ static void deprecated_set_speed(QTestState *who, long long value)
                           "'arguments': { 'value': %lld } }", value);
     g_assert(qdict_haskey(rsp, "return"));
     qobject_unref(rsp);
-    migrate_check_parameter(who, "max-bandwidth", value);
+    migrate_check_parameter_int(who, "max-bandwidth", value);
 }
 
 static void deprecated_set_cache_size(QTestState *who, long long value)
@@ -694,7 +695,7 @@ static void deprecated_set_cache_size(QTestState *who, long long value)
                          "'arguments': { 'value': %lld } }", value);
     g_assert(qdict_haskey(rsp, "return"));
     qobject_unref(rsp);
-    migrate_check_parameter(who, "xbzrle-cache-size", value);
+    migrate_check_parameter_int(who, "xbzrle-cache-size", value);
 }
 
 static void test_deprecated(void)
@@ -729,8 +730,8 @@ static int migrate_postcopy_prepare(QTestState **from_ptr,
      * quickly, but that it doesn't complete precopy even on a slow
      * machine, so also set the downtime.
      */
-    migrate_set_parameter(from, "max-bandwidth", 100000000);
-    migrate_set_parameter(from, "downtime-limit", 1);
+    migrate_set_parameter_int(from, "max-bandwidth", 100000000);
+    migrate_set_parameter_int(from, "downtime-limit", 1);
 
     /* Wait for the first serial output from the source */
     wait_for_serial("src_serial");
@@ -781,7 +782,7 @@ static void test_postcopy_recovery(void)
     }
 
     /* Turn postcopy speed down, 4K/s is slow enough on any machines */
-    migrate_set_parameter(from, "max-postcopy-bandwidth", 4096);
+    migrate_set_parameter_int(from, "max-postcopy-bandwidth", 4096);
 
     /* Now we start the postcopy */
     migrate_postcopy_start(from, to);
@@ -822,7 +823,7 @@ static void test_postcopy_recovery(void)
     g_free(uri);
 
     /* Restore the postcopy bandwidth to unlimited */
-    migrate_set_parameter(from, "max-postcopy-bandwidth", 0);
+    migrate_set_parameter_int(from, "max-postcopy-bandwidth", 0);
 
     migrate_postcopy_complete(from, to);
 }
@@ -868,9 +869,9 @@ static void test_precopy_unix(void)
      * machine, so also set the downtime.
      */
     /* 1 ms should make it not converge*/
-    migrate_set_parameter(from, "downtime-limit", 1);
+    migrate_set_parameter_int(from, "downtime-limit", 1);
     /* 1GB/s */
-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
 
     /* Wait for the first serial output from the source */
     wait_for_serial("src_serial");
@@ -880,7 +881,7 @@ static void test_precopy_unix(void)
     wait_for_migration_pass(from);
 
     /* 300 ms should converge */
-    migrate_set_parameter(from, "downtime-limit", 300);
+    migrate_set_parameter_int(from, "downtime-limit", 300);
 
     if (!got_stop) {
         qtest_qmp_eventwait(from, "STOP");
@@ -947,11 +948,11 @@ static void test_xbzrle(const char *uri)
      * machine, so also set the downtime.
      */
     /* 1 ms should make it not converge*/
-    migrate_set_parameter(from, "downtime-limit", 1);
+    migrate_set_parameter_int(from, "downtime-limit", 1);
     /* 1GB/s */
-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
 
-    migrate_set_parameter(from, "xbzrle-cache-size", 33554432);
+    migrate_set_parameter_int(from, "xbzrle-cache-size", 33554432);
 
     migrate_set_capability(from, "xbzrle", "true");
     migrate_set_capability(to, "xbzrle", "true");
@@ -963,7 +964,7 @@ static void test_xbzrle(const char *uri)
     wait_for_migration_pass(from);
 
     /* 300ms should converge */
-    migrate_set_parameter(from, "downtime-limit", 300);
+    migrate_set_parameter_int(from, "downtime-limit", 300);
 
     if (!got_stop) {
         qtest_qmp_eventwait(from, "STOP");
@@ -999,9 +1000,9 @@ static void test_precopy_tcp(void)
      * machine, so also set the downtime.
      */
     /* 1 ms should make it not converge*/
-    migrate_set_parameter(from, "downtime-limit", 1);
+    migrate_set_parameter_int(from, "downtime-limit", 1);
     /* 1GB/s */
-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
 
     /* Wait for the first serial output from the source */
     wait_for_serial("src_serial");
@@ -1013,7 +1014,7 @@ static void test_precopy_tcp(void)
     wait_for_migration_pass(from);
 
     /* 300ms should converge */
-    migrate_set_parameter(from, "downtime-limit", 300);
+    migrate_set_parameter_int(from, "downtime-limit", 300);
 
     if (!got_stop) {
         qtest_qmp_eventwait(from, "STOP");
-- 
2.21.0

[Qemu-devel] [PATCH v3 3/8] tests: Add migration multifd test

[Juan Quintela]

We set multifd-channels.

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

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 0b25aa3d6c..ff480e0682 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -1028,6 +1028,53 @@ static void test_precopy_tcp(void)
     g_free(uri);
 }
 
+static void test_multifd_tcp(void)
+{
+    char *uri;
+    QTestState *from, *to;
+
+    if (test_migrate_start(&from, &to, "tcp:127.0.0.1:0", false, false)) {
+        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", 2);
+    migrate_set_parameter_int(to, "multifd-channels", 2);
+
+    migrate_set_capability(from, "multifd", "true");
+    migrate_set_capability(to, "multifd", "true");
+    /* 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);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1082,6 +1129,7 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
+    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
 
     ret = g_test_run();
 
-- 
2.21.0

[Qemu-devel] [PATCH v3 4/8] 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 ff480e0682..65d5e256a7 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -427,6 +427,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.21.0

[Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Juan Quintela]

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

---
Rename it to NONE
Fix typos (dave)
---
 hmp.c                        | 17 +++++++++++++++++
 hw/core/qdev-properties.c    | 13 +++++++++++++
 include/hw/qdev-properties.h |  1 +
 migration/migration.c        | 16 ++++++++++++++++
 qapi/migration.json          | 30 +++++++++++++++++++++++++++---
 tests/migration-test.c       | 13 ++++++++++---
 6 files changed, 84 insertions(+), 6 deletions(-)

diff --git a/hmp.c b/hmp.c
index 56a3ed7375..5732c34249 100644
--- a/hmp.c
+++ b/hmp.c
@@ -38,6 +38,7 @@
 #include "qapi/qapi-commands-run-state.h"
 #include "qapi/qapi-commands-tpm.h"
 #include "qapi/qapi-commands-ui.h"
+#include "qapi/qapi-visit-migration.h"
 #include "qapi/qmp/qdict.h"
 #include "qapi/qmp/qerror.h"
 #include "qapi/string-input-visitor.h"
@@ -435,6 +436,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);
@@ -1736,6 +1740,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;
 
@@ -1821,6 +1826,18 @@ 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;
+        }
+        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {
+            error_setg(&err, "Invalid multifd_compress option %s", valuestr);
+            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/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
index 5da1439a8b..ebeeb5c88d 100644
--- a/hw/core/qdev-properties.c
+++ b/hw/core/qdev-properties.c
@@ -5,6 +5,7 @@
 #include "hw/pci/pci.h"
 #include "qapi/qmp/qerror.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"
@@ -645,6 +646,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 b6758c852e..ac452d8f2c 100644
--- a/include/hw/qdev-properties.h
+++ b/include/hw/qdev-properties.h
@@ -23,6 +23,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;
diff --git a/migration/migration.c b/migration/migration.c
index 609e0df5d0..d6f8ef342a 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -82,6 +82,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_COMPRESS MULTIFD_COMPRESS_NONE
 
 /* Background transfer rate for postcopy, 0 means unlimited, note
  * that page requests can still exceed this limit.
@@ -769,6 +770,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;
@@ -1268,6 +1271,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;
     }
@@ -1364,6 +1370,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);
@@ -3353,6 +3362,9 @@ void migration_global_dump(Monitor *mon)
 #define DEFINE_PROP_MIG_CAP(name, x)             \
     DEFINE_PROP_BOOL(name, MigrationState, enabled_capabilities[x], false)
 
+#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
+    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
+
 static Property migration_properties[] = {
     DEFINE_PROP_BOOL("store-global-state", MigrationState,
                      store_global_state, true),
@@ -3392,6 +3404,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),
@@ -3481,6 +3496,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/qapi/migration.json b/qapi/migration.json
index 9cfbaf8c6c..8ec1944b7a 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -482,6 +482,19 @@
 ##
 { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
 
+##
+# @MultifdCompress:
+#
+# An enumeration of multifd compression.
+#
+# @none: no compression.
+#
+# Since: 4.1
+#
+##
+{ 'enum': 'MultifdCompress',
+  'data': [ 'none' ] }
+
 ##
 # @MigrationParameter:
 #
@@ -580,6 +593,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',
@@ -592,7 +608,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:
@@ -682,6 +698,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
@@ -707,7 +726,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:
@@ -817,6 +837,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',
@@ -840,7 +863,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 65d5e256a7..8a1ccc2516 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -449,7 +449,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)
 {
@@ -1065,7 +1064,7 @@ static void test_precopy_tcp(void)
     g_free(uri);
 }
 
-static void test_multifd_tcp(void)
+static void test_multifd_tcp(const char *method)
 {
     char *uri;
     QTestState *from, *to;
@@ -1087,6 +1086,9 @@ static void test_multifd_tcp(void)
     migrate_set_parameter_int(from, "multifd-channels", 2);
     migrate_set_parameter_int(to, "multifd-channels", 2);
 
+    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");
     /* Wait for the first serial output from the source */
@@ -1112,6 +1114,11 @@ static void test_multifd_tcp(void)
     test_migrate_end(from, to, true);
 }
 
+static void test_multifd_tcp_none(void)
+{
+    test_multifd_tcp("none");
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -1166,7 +1173,7 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
-    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.21.0

[Qemu-devel] [PATCH v3 6/8] migration: Make none operations into its own structure

[Juan Quintela]

It will be used later.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 54 ++++++++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 49 insertions(+), 5 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index 1ca9ba77b6..6679e4f213 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -700,6 +700,40 @@ typedef struct {
     QemuSemaphore sem_sync;
 } MultiFDRecvParams;
 
+typedef struct {
+    /* Prepare the send packet */
+    int (*send_prepare)(MultiFDSendParams *p, uint32_t used);
+    /* Write the send packet */
+    int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **perr);
+    /* Read all pages */
+    int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **perr);
+} MultifdMethods;
+
+/* Multifd without compression */
+
+static int none_send_prepare(MultiFDSendParams *p, uint32_t used)
+{
+    p->next_packet_size = used * qemu_target_page_size();
+    return 0;
+}
+
+static int none_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
+{
+    return qio_channel_writev_all(p->c, p->pages->iov, used, perr);
+}
+
+static int none_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
+{
+    return qio_channel_readv_all(p->c, p->pages->iov, used, perr);
+
+}
+
+MultifdMethods multifd_none_ops = {
+    .send_prepare = none_send_prepare,
+    .send_write = none_send_write,
+    .recv_pages = none_recv_pages
+};
+
 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
 {
     MultiFDInit_t msg;
@@ -904,6 +938,8 @@ struct {
     uint64_t packet_num;
     /* send channels ready */
     QemuSemaphore channels_ready;
+    /* multifd ops */
+    MultifdMethods *ops;
 } *multifd_send_state;
 
 /*
@@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
     /* initial packet */
     p->num_packets = 1;
 
+    multifd_send_state->ops = &multifd_none_ops;
+
     while (true) {
         qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
@@ -1102,7 +1140,12 @@ static void *multifd_send_thread(void *opaque)
             uint64_t packet_num = p->packet_num;
             uint32_t flags = p->flags;
 
-            p->next_packet_size = used * qemu_target_page_size();
+            if (used) {
+                ret = multifd_send_state->ops->send_prepare(p, used);
+                if (ret != 0) {
+                    break;
+                }
+            }
             multifd_send_fill_packet(p);
             p->flags = 0;
             p->num_packets++;
@@ -1120,8 +1163,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;
                 }
@@ -1223,6 +1265,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)
@@ -1324,6 +1368,7 @@ static void *multifd_recv_thread(void *opaque)
     trace_multifd_recv_thread_start(p->id);
     rcu_register_thread();
 
+    multifd_recv_state->ops = &multifd_none_ops;
     while (true) {
         uint32_t used;
         uint32_t flags;
@@ -1353,8 +1398,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;
             }
-- 
2.21.0

[Qemu-devel] [PATCH v3 7/8] multifd: Add zlib compression support

[Juan Quintela]

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

diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
index ebeeb5c88d..e40aa806e2 100644
--- a/hw/core/qdev-properties.c
+++ b/hw/core/qdev-properties.c
@@ -651,7 +651,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/migration.c b/migration/migration.c
index d6f8ef342a..69d85cbe5e 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -2141,6 +2141,15 @@ bool migrate_use_multifd(void)
     return s->enabled_capabilities[MIGRATION_CAPABILITY_MULTIFD];
 }
 
+bool migrate_use_multifd_zlib(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->parameters.multifd_compress == MULTIFD_COMPRESS_ZLIB;
+}
+
 bool migrate_pause_before_switchover(void)
 {
     MigrationState *s;
diff --git a/migration/migration.h b/migration/migration.h
index 438f17edad..fc4fb841d4 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -269,6 +269,7 @@ bool migrate_ignore_shared(void);
 
 bool migrate_auto_converge(void);
 bool migrate_use_multifd(void);
+bool migrate_use_multifd_zlib(void);
 bool migrate_pause_before_switchover(void);
 int migrate_multifd_channels(void);
 
diff --git a/migration/ram.c b/migration/ram.c
index 6679e4f213..fdb5bf07a5 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -582,6 +582,7 @@ exit:
 #define MULTIFD_VERSION 1
 
 #define MULTIFD_FLAG_SYNC (1 << 0)
+#define MULTIFD_FLAG_ZLIB (1 << 1)
 
 /* This value needs to be a multiple of qemu_target_page_size() */
 #define MULTIFD_PACKET_SIZE (512 * 1024)
@@ -663,6 +664,12 @@ typedef struct {
     uint64_t num_pages;
     /* syncs main thread and channels */
     QemuSemaphore sem_sync;
+    /* stream for compression */
+    z_stream zs;
+    /* compressed buffer */
+    uint8_t *zbuff;
+    /* size of compressed buffer */
+    uint32_t zbuff_len;
 }  MultiFDSendParams;
 
 typedef struct {
@@ -698,6 +705,12 @@ typedef struct {
     uint64_t num_pages;
     /* syncs main thread and channels */
     QemuSemaphore sem_sync;
+    /* stream for compression */
+    z_stream zs;
+    /* compressed buffer */
+    uint8_t *zbuff;
+    /* size of compressed buffer */
+    uint32_t zbuff_len;
 } MultiFDRecvParams;
 
 typedef struct {
@@ -1071,6 +1084,9 @@ void multifd_save_cleanup(void)
         p->packet_len = 0;
         g_free(p->packet);
         p->packet = NULL;
+        deflateEnd(&p->zs);
+        g_free(p->zbuff);
+        p->zbuff = NULL;
     }
     qemu_sem_destroy(&multifd_send_state->channels_ready);
     qemu_sem_destroy(&multifd_send_state->sem_sync);
@@ -1240,6 +1256,7 @@ int multifd_save_setup(void)
 
     for (i = 0; i < thread_count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
+        z_stream *zs = &p->zs;
 
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem, 0);
@@ -1253,6 +1270,17 @@ int multifd_save_setup(void)
         p->packet = g_malloc0(p->packet_len);
         p->name = g_strdup_printf("multifdsend_%d", i);
         socket_send_channel_create(multifd_new_send_channel_async, p);
+        zs->zalloc = Z_NULL;
+        zs->zfree = Z_NULL;
+        zs->opaque = Z_NULL;
+        if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
+            printf("deflate init failed\n");
+            return -1;
+        }
+        /* We will never have more than page_count pages */
+        p->zbuff_len = page_count * qemu_target_page_size();
+        p->zbuff_len *= 2;
+        p->zbuff = g_malloc0(p->zbuff_len);
     }
     return 0;
 }
@@ -1322,6 +1350,9 @@ int multifd_load_cleanup(Error **errp)
         p->packet_len = 0;
         g_free(p->packet);
         p->packet = NULL;
+        inflateEnd(&p->zs);
+        g_free(p->zbuff);
+        p->zbuff = NULL;
     }
     qemu_sem_destroy(&multifd_recv_state->sem_sync);
     g_free(multifd_recv_state->params);
@@ -1440,6 +1471,7 @@ int multifd_load_setup(void)
 
     for (i = 0; i < thread_count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
+        z_stream *zs = &p->zs;
 
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem_sync, 0);
@@ -1449,6 +1481,21 @@ int multifd_load_setup(void)
                       + sizeof(ram_addr_t) * page_count;
         p->packet = g_malloc0(p->packet_len);
         p->name = g_strdup_printf("multifdrecv_%d", i);
+
+        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) {
+            printf("inflate init failed\n");
+            return -1;
+        }
+        /* We will never have more than page_count pages */
+        p->zbuff_len = page_count * qemu_target_page_size();
+        /* We know compression "could" use more space */
+        p->zbuff_len *= 2;
+        p->zbuff = g_malloc0(p->zbuff_len);
     }
     return 0;
 }
diff --git a/qapi/migration.json b/qapi/migration.json
index 8ec1944b7a..e6c27fae06 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -493,7 +493,7 @@
 #
 ##
 { 'enum': 'MultifdCompress',
-  'data': [ 'none' ] }
+  'data': [ 'none', 'zlib' ] }
 
 ##
 # @MigrationParameter:
diff --git a/tests/migration-test.c b/tests/migration-test.c
index 8a1ccc2516..2dd4d4c5b4 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -1119,6 +1119,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";
@@ -1174,6 +1179,7 @@ int main(int argc, char **argv)
     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
     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.21.0

[Qemu-devel] [PATCH v3 8/8] multifd: rest of zlib compression

[Juan Quintela]

This is still a work in progress, but get everything sent as expected
and it is faster than the code that is already there.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 106 +++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 104 insertions(+), 2 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index fdb5bf07a5..efbb253c1a 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -747,6 +747,100 @@ MultifdMethods multifd_none_ops = {
     .recv_pages = none_recv_pages
 };
 
+/* Multifd zlib compression */
+
+static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used)
+{
+    struct iovec *iov = p->pages->iov;
+    z_stream *zs = &p->zs;
+    uint32_t out_size = 0;
+    int ret;
+    int i;
+
+    for (i = 0; i < used; i++) {
+        uint32_t available = p->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 = p->zbuff + out_size;
+
+        ret = deflate(zs, flush);
+        if (ret != Z_OK) {
+            printf("problem with deflate? %d\n", ret);
+            qemu_mutex_unlock(&p->mutex);
+            return -1;
+        }
+        out_size += available - zs->avail_out;
+    }
+    p->next_packet_size = out_size;
+
+    return 0;
+}
+
+static int zlib_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
+{
+    return qio_channel_write_all(p->c, (void *)p->zbuff, p->next_packet_size,
+                                 perr);
+}
+
+static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
+{
+    uint32_t in_size = p->next_packet_size;
+    uint32_t out_size = 0;
+    uint32_t expected_size = used * qemu_target_page_size();
+    z_stream *zs = &p->zs;
+    int ret;
+    int i;
+
+    ret = qio_channel_read_all(p->c, (void *)p->zbuff, in_size, perr);
+
+    if (ret != 0) {
+        return ret;
+    }
+
+    zs->avail_in = in_size;
+    zs->next_in = p->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) {
+            printf("%d: problem with inflate? %d\n", p->id, ret);
+            qemu_mutex_unlock(&p->mutex);
+            return ret;
+        }
+        out_size += iov->iov_len;
+    }
+    if (out_size != expected_size) {
+        printf("out size %d expected size %d\n",
+               out_size, expected_size);
+        return -1;
+    }
+    return 0;
+}
+
+MultifdMethods multifd_zlib_ops = {
+    .send_prepare = zlib_send_prepare,
+    .send_write = zlib_send_write,
+    .recv_pages = zlib_recv_pages
+};
+
 static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
 {
     MultiFDInit_t msg;
@@ -1145,7 +1239,11 @@ static void *multifd_send_thread(void *opaque)
     /* initial packet */
     p->num_packets = 1;
 
-    multifd_send_state->ops = &multifd_none_ops;
+    if (migrate_use_multifd_zlib()) {
+        multifd_send_state->ops = &multifd_zlib_ops;
+    } else {
+        multifd_send_state->ops = &multifd_none_ops;
+    }
 
     while (true) {
         qemu_sem_wait(&p->sem);
@@ -1399,7 +1497,11 @@ static void *multifd_recv_thread(void *opaque)
     trace_multifd_recv_thread_start(p->id);
     rcu_register_thread();
 
-    multifd_recv_state->ops = &multifd_none_ops;
+    if (migrate_use_multifd_zlib()) {
+        multifd_recv_state->ops = &multifd_zlib_ops;
+    } else {
+        multifd_recv_state->ops = &multifd_none_ops;
+    }
     while (true) {
         uint32_t used;
         uint32_t flags;
-- 
2.21.0

Re: [Qemu-devel] [PATCH v3 5/8] 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)
> ---
>  hmp.c                        | 17 +++++++++++++++++
>  hw/core/qdev-properties.c    | 13 +++++++++++++
>  include/hw/qdev-properties.h |  1 +
>  migration/migration.c        | 16 ++++++++++++++++
>  qapi/migration.json          | 30 +++++++++++++++++++++++++++---
>  tests/migration-test.c       | 13 ++++++++++---
>  6 files changed, 84 insertions(+), 6 deletions(-)
>
> diff --git a/hmp.c b/hmp.c
> index 56a3ed7375..5732c34249 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -38,6 +38,7 @@
>  #include "qapi/qapi-commands-run-state.h"
>  #include "qapi/qapi-commands-tpm.h"
>  #include "qapi/qapi-commands-ui.h"
> +#include "qapi/qapi-visit-migration.h"
>  #include "qapi/qmp/qdict.h"
>  #include "qapi/qmp/qerror.h"
>  #include "qapi/string-input-visitor.h"
> @@ -435,6 +436,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);
> @@ -1736,6 +1740,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;
>  
> @@ -1821,6 +1826,18 @@ 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;
> +        }
> +        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {
> +            error_setg(&err, "Invalid multifd_compress option %s", valuestr);
> +            break;
> +        }

This should never happen.  If you want to check anyway, make it an
assertion.

Just in case you don't believe me, or are curious:

visit_type_MultifdCompress() wraps around visit_type_enum(), passing it
&MultifdCompress_lookup.

Since @v is an input visitor, visit_type_enum() wraps around
input_type_enum().

input_type_enum() computes the value to store in @compress_type with
qapi_enum_parse().

To get here, visit_type_MultifdCompress() must have succeeded,
i.e. visit_type_enum(), input_type_enum() and qapi_enum_parse() all
succeded.

On success, qapi_enum_parse() returns one of the values in
MultifdCompress_lookup, i.e. a member of enum MultifdCompress other than
MULTIFD_COMPRESS__MAX.

> +        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/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
> index 5da1439a8b..ebeeb5c88d 100644
> --- a/hw/core/qdev-properties.c
> +++ b/hw/core/qdev-properties.c
> @@ -5,6 +5,7 @@
>  #include "hw/pci/pci.h"
>  #include "qapi/qmp/qerror.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"
> @@ -645,6 +646,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",

This looks weird now, but it'll make sense when PATCH 7 adds the second
value.

> +    .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 b6758c852e..ac452d8f2c 100644
> --- a/include/hw/qdev-properties.h
> +++ b/include/hw/qdev-properties.h
> @@ -23,6 +23,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;
> diff --git a/migration/migration.c b/migration/migration.c
> index 609e0df5d0..d6f8ef342a 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -82,6 +82,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_COMPRESS MULTIFD_COMPRESS_NONE
>  
>  /* Background transfer rate for postcopy, 0 means unlimited, note
>   * that page requests can still exceed this limit.
> @@ -769,6 +770,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;
> @@ -1268,6 +1271,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;
>      }
> @@ -1364,6 +1370,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);
> @@ -3353,6 +3362,9 @@ void migration_global_dump(Monitor *mon)
>  #define DEFINE_PROP_MIG_CAP(name, x)             \
>      DEFINE_PROP_BOOL(name, MigrationState, enabled_capabilities[x], false)
>  
> +#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
> +    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
> +

Did you forget to move this?

>  static Property migration_properties[] = {
>      DEFINE_PROP_BOOL("store-global-state", MigrationState,
>                       store_global_state, true),
> @@ -3392,6 +3404,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),
> @@ -3481,6 +3496,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/qapi/migration.json b/qapi/migration.json
> index 9cfbaf8c6c..8ec1944b7a 100644
> --- a/qapi/migration.json
> +++ b/qapi/migration.json
> @@ -482,6 +482,19 @@
>  ##
>  { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
>  
> +##
> +# @MultifdCompress:
> +#
> +# An enumeration of multifd compression.
> +#
> +# @none: no compression.
> +#
> +# Since: 4.1
> +#
> +##
> +{ 'enum': 'MultifdCompress',
> +  'data': [ 'none' ] }
> +
>  ##
>  # @MigrationParameter:
>  #
> @@ -580,6 +593,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',
> @@ -592,7 +608,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:
> @@ -682,6 +698,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
> @@ -707,7 +726,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:
> @@ -817,6 +837,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',
> @@ -840,7 +863,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:

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

[...]

Re: [Qemu-devel] [PATCH v3 1/8] migration: fix multifd_recv event typo

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> It uses num in multifd_send().  Make it coherent.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>

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

> ---
>  migration/trace-events | 2 +-
>  1 file changed, 1 insertion(+), 1 deletion(-)
> 
> diff --git a/migration/trace-events b/migration/trace-events
> index de2e136e57..cd50a1e659 100644
> --- a/migration/trace-events
> +++ b/migration/trace-events
> @@ -80,7 +80,7 @@ get_queued_page_not_dirty(const char *block_name, uint64_t tmp_offset, unsigned
>  migration_bitmap_sync_start(void) ""
>  migration_bitmap_sync_end(uint64_t dirty_pages) "dirty_pages %" PRIu64
>  migration_throttle(void) ""
> -multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet number %" PRIu64 " pages %d flags 0x%x next packet size %d"
> +multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet_num %" PRIu64 " pages %d flags 0x%x next packet size %d"
>  multifd_recv_sync_main(long packet_num) "packet num %ld"
>  multifd_recv_sync_main_signal(uint8_t id) "channel %d"
>  multifd_recv_sync_main_wait(uint8_t id) "channel %d"
> -- 
> 2.21.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v3 3/8] tests: Add migration multifd test

[Wei Yang]

On Wed, May 15, 2019 at 02:15:39PM +0200, Juan Quintela wrote:
>We set multifd-channels.
>
>Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>Reviewed-by: Thomas Huth <thuth@redhat.com>
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>---
> tests/migration-test.c | 48 ++++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 48 insertions(+)
>
>diff --git a/tests/migration-test.c b/tests/migration-test.c
>index 0b25aa3d6c..ff480e0682 100644
>--- a/tests/migration-test.c
>+++ b/tests/migration-test.c
>@@ -1028,6 +1028,53 @@ static void test_precopy_tcp(void)
>     g_free(uri);
> }
> 
>+static void test_multifd_tcp(void)
>+{
>+    char *uri;
>+    QTestState *from, *to;
>+
>+    if (test_migrate_start(&from, &to, "tcp:127.0.0.1:0", false, false)) {
>+        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", 2);
>+    migrate_set_parameter_int(to, "multifd-channels", 2);
>+
>+    migrate_set_capability(from, "multifd", "true");
>+    migrate_set_capability(to, "multifd", "true");

Hi, Juan

In case to test multifd, what's the minimum configuration? Set multifd
capability at both side is fine?

>+    /* 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);
>+}
>+
> int main(int argc, char **argv)
> {
>     char template[] = "/tmp/migration-test-XXXXXX";
>@@ -1082,6 +1129,7 @@ int main(int argc, char **argv)
>     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>+    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
> 
>     ret = g_test_run();
> 
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 0/8] WIP: Multifd compression support

[Wei Yang]

On Wed, May 15, 2019 at 02:15:36PM +0200, Juan Quintela wrote:
>v3:
>- improve the code
>- address David and Markus comments
>- make compression code into methods
>  so we can add any other method ading just three functions
>
>Please review, as far as I know everything is ok now.
>
>Todo: Add zstd support

Confusion here. It is zstd or sztd?

BTW, I am not sure what it is :-)

>
>v2:
>- improve the code left and right
>- Split better the zlib code
>- rename everything to v4.1
>- Add tests for multifd-compress zlib
>- Parameter is now an enum (soon will see sztd)
                                           ^^^
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 1/8] migration: fix multifd_recv event typo

[Wei Yang]

On Wed, May 15, 2019 at 02:15:37PM +0200, Juan Quintela wrote:
>It uses num in multifd_send().  Make it coherent.
>
>Signed-off-by: Juan Quintela <quintela@redhat.com>

Reviewed-by: Wei Yang <richardw.yang@linux.intel.com>

>---
> migration/trace-events | 2 +-
> 1 file changed, 1 insertion(+), 1 deletion(-)
>
>diff --git a/migration/trace-events b/migration/trace-events
>index de2e136e57..cd50a1e659 100644
>--- a/migration/trace-events
>+++ b/migration/trace-events
>@@ -80,7 +80,7 @@ get_queued_page_not_dirty(const char *block_name, uint64_t tmp_offset, unsigned
> migration_bitmap_sync_start(void) ""
> migration_bitmap_sync_end(uint64_t dirty_pages) "dirty_pages %" PRIu64
> migration_throttle(void) ""
>-multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet number %" PRIu64 " pages %d flags 0x%x next packet size %d"
>+multifd_recv(uint8_t id, uint64_t packet_num, uint32_t used, uint32_t flags, uint32_t next_packet_size) "channel %d packet_num %" PRIu64 " pages %d flags 0x%x next packet size %d"
> multifd_recv_sync_main(long packet_num) "packet num %ld"
> multifd_recv_sync_main_signal(uint8_t id) "channel %d"
> multifd_recv_sync_main_wait(uint8_t id) "channel %d"
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 2/8] migration-test: rename parameter to parameter_int

[Wei Yang]

On Wed, May 15, 2019 at 02:15:38PM +0200, Juan Quintela wrote:
>We would need _str ones on the next patch.
>
>Signed-off-by: Juan Quintela <quintela@redhat.com>

Reviewed-by: Wei Yang <richardw.yang@linux.intel.com>

>---
> tests/migration-test.c | 49 +++++++++++++++++++++---------------------
> 1 file changed, 25 insertions(+), 24 deletions(-)
>
>diff --git a/tests/migration-test.c b/tests/migration-test.c
>index bd3f5c3125..0b25aa3d6c 100644
>--- a/tests/migration-test.c
>+++ b/tests/migration-test.c
>@@ -392,7 +392,8 @@ static char *migrate_get_socket_address(QTestState *who, const char *parameter)
>     return result;
> }
> 
>-static long long migrate_get_parameter(QTestState *who, const char *parameter)
>+static long long migrate_get_parameter_int(QTestState *who,
>+                                           const char *parameter)
> {
>     QDict *rsp;
>     long long result;
>@@ -403,17 +404,17 @@ static long long migrate_get_parameter(QTestState *who, const char *parameter)
>     return result;
> }
> 
>-static void migrate_check_parameter(QTestState *who, const char *parameter,
>-                                    long long value)
>+static void migrate_check_parameter_int(QTestState *who, const char *parameter,
>+                                        long long value)
> {
>     long long result;
> 
>-    result = migrate_get_parameter(who, parameter);
>+    result = migrate_get_parameter_int(who, parameter);
>     g_assert_cmpint(result, ==, value);
> }
> 
>-static void migrate_set_parameter(QTestState *who, const char *parameter,
>-                                  long long value)
>+static void migrate_set_parameter_int(QTestState *who, const char *parameter,
>+                                      long long value)
> {
>     QDict *rsp;
> 
>@@ -423,7 +424,7 @@ static void migrate_set_parameter(QTestState *who, const char *parameter,
>                     parameter, value);
>     g_assert(qdict_haskey(rsp, "return"));
>     qobject_unref(rsp);
>-    migrate_check_parameter(who, parameter, value);
>+    migrate_check_parameter_int(who, parameter, value);
> }
> 
> static void migrate_pause(QTestState *who)
>@@ -672,7 +673,7 @@ static void deprecated_set_downtime(QTestState *who, const double value)
>                     " 'arguments': { 'value': %f } }", value);
>     g_assert(qdict_haskey(rsp, "return"));
>     qobject_unref(rsp);
>-    migrate_check_parameter(who, "downtime-limit", value * 1000);
>+    migrate_check_parameter_int(who, "downtime-limit", value * 1000);
> }
> 
> static void deprecated_set_speed(QTestState *who, long long value)
>@@ -683,7 +684,7 @@ static void deprecated_set_speed(QTestState *who, long long value)
>                           "'arguments': { 'value': %lld } }", value);
>     g_assert(qdict_haskey(rsp, "return"));
>     qobject_unref(rsp);
>-    migrate_check_parameter(who, "max-bandwidth", value);
>+    migrate_check_parameter_int(who, "max-bandwidth", value);
> }
> 
> static void deprecated_set_cache_size(QTestState *who, long long value)
>@@ -694,7 +695,7 @@ static void deprecated_set_cache_size(QTestState *who, long long value)
>                          "'arguments': { 'value': %lld } }", value);
>     g_assert(qdict_haskey(rsp, "return"));
>     qobject_unref(rsp);
>-    migrate_check_parameter(who, "xbzrle-cache-size", value);
>+    migrate_check_parameter_int(who, "xbzrle-cache-size", value);
> }
> 
> static void test_deprecated(void)
>@@ -729,8 +730,8 @@ static int migrate_postcopy_prepare(QTestState **from_ptr,
>      * quickly, but that it doesn't complete precopy even on a slow
>      * machine, so also set the downtime.
>      */
>-    migrate_set_parameter(from, "max-bandwidth", 100000000);
>-    migrate_set_parameter(from, "downtime-limit", 1);
>+    migrate_set_parameter_int(from, "max-bandwidth", 100000000);
>+    migrate_set_parameter_int(from, "downtime-limit", 1);
> 
>     /* Wait for the first serial output from the source */
>     wait_for_serial("src_serial");
>@@ -781,7 +782,7 @@ static void test_postcopy_recovery(void)
>     }
> 
>     /* Turn postcopy speed down, 4K/s is slow enough on any machines */
>-    migrate_set_parameter(from, "max-postcopy-bandwidth", 4096);
>+    migrate_set_parameter_int(from, "max-postcopy-bandwidth", 4096);
> 
>     /* Now we start the postcopy */
>     migrate_postcopy_start(from, to);
>@@ -822,7 +823,7 @@ static void test_postcopy_recovery(void)
>     g_free(uri);
> 
>     /* Restore the postcopy bandwidth to unlimited */
>-    migrate_set_parameter(from, "max-postcopy-bandwidth", 0);
>+    migrate_set_parameter_int(from, "max-postcopy-bandwidth", 0);
> 
>     migrate_postcopy_complete(from, to);
> }
>@@ -868,9 +869,9 @@ static void test_precopy_unix(void)
>      * machine, so also set the downtime.
>      */
>     /* 1 ms should make it not converge*/
>-    migrate_set_parameter(from, "downtime-limit", 1);
>+    migrate_set_parameter_int(from, "downtime-limit", 1);
>     /* 1GB/s */
>-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
>+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
> 
>     /* Wait for the first serial output from the source */
>     wait_for_serial("src_serial");
>@@ -880,7 +881,7 @@ static void test_precopy_unix(void)
>     wait_for_migration_pass(from);
> 
>     /* 300 ms should converge */
>-    migrate_set_parameter(from, "downtime-limit", 300);
>+    migrate_set_parameter_int(from, "downtime-limit", 300);
> 
>     if (!got_stop) {
>         qtest_qmp_eventwait(from, "STOP");
>@@ -947,11 +948,11 @@ static void test_xbzrle(const char *uri)
>      * machine, so also set the downtime.
>      */
>     /* 1 ms should make it not converge*/
>-    migrate_set_parameter(from, "downtime-limit", 1);
>+    migrate_set_parameter_int(from, "downtime-limit", 1);
>     /* 1GB/s */
>-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
>+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
> 
>-    migrate_set_parameter(from, "xbzrle-cache-size", 33554432);
>+    migrate_set_parameter_int(from, "xbzrle-cache-size", 33554432);
> 
>     migrate_set_capability(from, "xbzrle", "true");
>     migrate_set_capability(to, "xbzrle", "true");
>@@ -963,7 +964,7 @@ static void test_xbzrle(const char *uri)
>     wait_for_migration_pass(from);
> 
>     /* 300ms should converge */
>-    migrate_set_parameter(from, "downtime-limit", 300);
>+    migrate_set_parameter_int(from, "downtime-limit", 300);
> 
>     if (!got_stop) {
>         qtest_qmp_eventwait(from, "STOP");
>@@ -999,9 +1000,9 @@ static void test_precopy_tcp(void)
>      * machine, so also set the downtime.
>      */
>     /* 1 ms should make it not converge*/
>-    migrate_set_parameter(from, "downtime-limit", 1);
>+    migrate_set_parameter_int(from, "downtime-limit", 1);
>     /* 1GB/s */
>-    migrate_set_parameter(from, "max-bandwidth", 1000000000);
>+    migrate_set_parameter_int(from, "max-bandwidth", 1000000000);
> 
>     /* Wait for the first serial output from the source */
>     wait_for_serial("src_serial");
>@@ -1013,7 +1014,7 @@ static void test_precopy_tcp(void)
>     wait_for_migration_pass(from);
> 
>     /* 300ms should converge */
>-    migrate_set_parameter(from, "downtime-limit", 300);
>+    migrate_set_parameter_int(from, "downtime-limit", 300);
> 
>     if (!got_stop) {
>         qtest_qmp_eventwait(from, "STOP");
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 3/8] tests: Add migration multifd test

[Wei Yang]

On Wed, May 15, 2019 at 02:15:39PM +0200, Juan Quintela wrote:
>We set multifd-channels.
>
>Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>Reviewed-by: Thomas Huth <thuth@redhat.com>
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>---
> tests/migration-test.c | 48 ++++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 48 insertions(+)
>
>diff --git a/tests/migration-test.c b/tests/migration-test.c
>index 0b25aa3d6c..ff480e0682 100644
>--- a/tests/migration-test.c
>+++ b/tests/migration-test.c
>@@ -1028,6 +1028,53 @@ static void test_precopy_tcp(void)
>     g_free(uri);
> }
> 
>+static void test_multifd_tcp(void)
>+{
>+    char *uri;
>+    QTestState *from, *to;
>+
>+    if (test_migrate_start(&from, &to, "tcp:127.0.0.1:0", false, false)) {
>+        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", 2);
>+    migrate_set_parameter_int(to, "multifd-channels", 2);
>+
>+    migrate_set_capability(from, "multifd", "true");
>+    migrate_set_capability(to, "multifd", "true");
>+    /* 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);

Miss free uri here?

    g_free(uri);

And

Tested-by: Wei Yang <richardw.yang@linux.intel.com>

>+}
>+
> int main(int argc, char **argv)
> {
>     char template[] = "/tmp/migration-test-XXXXXX";
>@@ -1082,6 +1129,7 @@ int main(int argc, char **argv)
>     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>+    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
> 
>     ret = g_test_run();
> 
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Wei Yang]

On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>
>---
>Rename it to NONE
>Fix typos (dave)
>---
> hmp.c                        | 17 +++++++++++++++++
> hw/core/qdev-properties.c    | 13 +++++++++++++
> include/hw/qdev-properties.h |  1 +
> migration/migration.c        | 16 ++++++++++++++++
> qapi/migration.json          | 30 +++++++++++++++++++++++++++---
> tests/migration-test.c       | 13 ++++++++++---
> 6 files changed, 84 insertions(+), 6 deletions(-)
>
>diff --git a/hmp.c b/hmp.c
>index 56a3ed7375..5732c34249 100644
>--- a/hmp.c
>+++ b/hmp.c
>@@ -38,6 +38,7 @@
> #include "qapi/qapi-commands-run-state.h"
> #include "qapi/qapi-commands-tpm.h"
> #include "qapi/qapi-commands-ui.h"
>+#include "qapi/qapi-visit-migration.h"
> #include "qapi/qmp/qdict.h"
> #include "qapi/qmp/qerror.h"
> #include "qapi/string-input-visitor.h"
>@@ -435,6 +436,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);
>@@ -1736,6 +1740,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;
> 
>@@ -1821,6 +1826,18 @@ 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;
>+        }
>+        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {
>+            error_setg(&err, "Invalid multifd_compress option %s", valuestr);
>+            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/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
>index 5da1439a8b..ebeeb5c88d 100644
>--- a/hw/core/qdev-properties.c
>+++ b/hw/core/qdev-properties.c
>@@ -5,6 +5,7 @@
> #include "hw/pci/pci.h"
> #include "qapi/qmp/qerror.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"
>@@ -645,6 +646,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 b6758c852e..ac452d8f2c 100644
>--- a/include/hw/qdev-properties.h
>+++ b/include/hw/qdev-properties.h
>@@ -23,6 +23,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;
>diff --git a/migration/migration.c b/migration/migration.c
>index 609e0df5d0..d6f8ef342a 100644
>--- a/migration/migration.c
>+++ b/migration/migration.c
>@@ -82,6 +82,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_COMPRESS MULTIFD_COMPRESS_NONE
> 
> /* Background transfer rate for postcopy, 0 means unlimited, note
>  * that page requests can still exceed this limit.
>@@ -769,6 +770,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;
>@@ -1268,6 +1271,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;
>     }
>@@ -1364,6 +1370,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);
>@@ -3353,6 +3362,9 @@ void migration_global_dump(Monitor *mon)
> #define DEFINE_PROP_MIG_CAP(name, x)             \
>     DEFINE_PROP_BOOL(name, MigrationState, enabled_capabilities[x], false)
> 
>+#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
>+    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
>+
> static Property migration_properties[] = {
>     DEFINE_PROP_BOOL("store-global-state", MigrationState,
>                      store_global_state, true),
>@@ -3392,6 +3404,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),
>@@ -3481,6 +3496,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/qapi/migration.json b/qapi/migration.json
>index 9cfbaf8c6c..8ec1944b7a 100644
>--- a/qapi/migration.json
>+++ b/qapi/migration.json
>@@ -482,6 +482,19 @@
> ##
> { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
> 
>+##
>+# @MultifdCompress:
>+#
>+# An enumeration of multifd compression.
>+#
>+# @none: no compression.
>+#
>+# Since: 4.1
>+#
>+##
>+{ 'enum': 'MultifdCompress',
>+  'data': [ 'none' ] }
>+
> ##
> # @MigrationParameter:
> #
>@@ -580,6 +593,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',
>@@ -592,7 +608,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:
>@@ -682,6 +698,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
>@@ -707,7 +726,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:
>@@ -817,6 +837,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',
>@@ -840,7 +863,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 65d5e256a7..8a1ccc2516 100644
>--- a/tests/migration-test.c
>+++ b/tests/migration-test.c

Well, may I suggest to split the test into another one?

>@@ -449,7 +449,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)
> {
>@@ -1065,7 +1064,7 @@ static void test_precopy_tcp(void)
>     g_free(uri);
> }
> 
>-static void test_multifd_tcp(void)
>+static void test_multifd_tcp(const char *method)
> {
>     char *uri;
>     QTestState *from, *to;
>@@ -1087,6 +1086,9 @@ static void test_multifd_tcp(void)
>     migrate_set_parameter_int(from, "multifd-channels", 2);
>     migrate_set_parameter_int(to, "multifd-channels", 2);
> 
>+    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");
>     /* Wait for the first serial output from the source */
>@@ -1112,6 +1114,11 @@ static void test_multifd_tcp(void)
>     test_migrate_end(from, to, true);
> }
> 
>+static void test_multifd_tcp_none(void)
>+{
>+    test_multifd_tcp("none");
>+}
>+
> int main(int argc, char **argv)
> {
>     char template[] = "/tmp/migration-test-XXXXXX";
>@@ -1166,7 +1173,7 @@ int main(int argc, char **argv)
>     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>-    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.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Wei Yang]

On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>
>---
>Rename it to NONE
>Fix typos (dave)
>---
> hmp.c                        | 17 +++++++++++++++++
> hw/core/qdev-properties.c    | 13 +++++++++++++
> include/hw/qdev-properties.h |  1 +
> migration/migration.c        | 16 ++++++++++++++++
> qapi/migration.json          | 30 +++++++++++++++++++++++++++---
> tests/migration-test.c       | 13 ++++++++++---
> 6 files changed, 84 insertions(+), 6 deletions(-)
>
>diff --git a/hmp.c b/hmp.c
>index 56a3ed7375..5732c34249 100644
>--- a/hmp.c
>+++ b/hmp.c
>@@ -38,6 +38,7 @@
> #include "qapi/qapi-commands-run-state.h"
> #include "qapi/qapi-commands-tpm.h"
> #include "qapi/qapi-commands-ui.h"
>+#include "qapi/qapi-visit-migration.h"
> #include "qapi/qmp/qdict.h"
> #include "qapi/qmp/qerror.h"
> #include "qapi/string-input-visitor.h"
>@@ -435,6 +436,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);
>@@ -1736,6 +1740,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;
> 
>@@ -1821,6 +1826,18 @@ 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;
>+        }
>+        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {

A warning during build:

hmp.c:1835:27: warning: comparison of unsigned enum expression < 0 is always false [-Wtautological-compare]
        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Wei Yang]

On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>diff --git a/qapi/migration.json b/qapi/migration.json
>index 9cfbaf8c6c..8ec1944b7a 100644
>--- a/qapi/migration.json
>+++ b/qapi/migration.json
>@@ -482,6 +482,19 @@
> ##
> { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
> 
>+##
>+# @MultifdCompress:
>+#
>+# An enumeration of multifd compression.
>+#
>+# @none: no compression.
>+#
>+# Since: 4.1
>+#
>+##
>+{ 'enum': 'MultifdCompress',
>+  'data': [ 'none' ] }
>+
> ##
> # @MigrationParameter:
> #
>@@ -580,6 +593,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',
>@@ -592,7 +608,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:
>@@ -682,6 +698,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
>@@ -707,7 +726,8 @@
>             '*multifd-channels': 'int',
>             '*xbzrle-cache-size': 'size',
>             '*max-postcopy-bandwidth': 'size',
>-	    '*max-cpu-throttle': 'int' } }
>+	    '*max-cpu-throttle': 'int',

A tab at the beginning, it would be better to fix this :-)

>+            '*multifd-compress': 'MultifdCompress' } }
> 
> ##
> # @migrate-set-parameters:
>@@ -817,6 +837,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',
>@@ -840,7 +863,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:

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 6/8] migration: Make none operations into its own structure

[Wei Yang]

On Wed, May 15, 2019 at 02:15:42PM +0200, Juan Quintela wrote:
>+
>+MultifdMethods multifd_none_ops = {
>+    .send_prepare = none_send_prepare,
>+    .send_write = none_send_write,
>+    .recv_pages = none_recv_pages
>+};
>+
> static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
> {
>     MultiFDInit_t msg;
>@@ -904,6 +938,8 @@ struct {
>     uint64_t packet_num;
>     /* send channels ready */
>     QemuSemaphore channels_ready;
>+    /* multifd ops */
>+    MultifdMethods *ops;
> } *multifd_send_state;
> 
> /*
>@@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
>     /* initial packet */
>     p->num_packets = 1;
> 
>+    multifd_send_state->ops = &multifd_none_ops;
>+

I am afraid it is not a good practice to assign ops when each thread starts
work.

>     while (true) {
>         qemu_sem_wait(&p->sem);
>         qemu_mutex_lock(&p->mutex);
>@@ -1102,7 +1140,12 @@ static void *multifd_send_thread(void *opaque)
>             uint64_t packet_num = p->packet_num;
>             uint32_t flags = p->flags;
> 
>-            p->next_packet_size = used * qemu_target_page_size();
>+            if (used) {
>+                ret = multifd_send_state->ops->send_prepare(p, used);
>+                if (ret != 0) {
>+                    break;
>+                }
>+            }
>             multifd_send_fill_packet(p);
>             p->flags = 0;
>             p->num_packets++;
>@@ -1120,8 +1163,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;
>                 }
>@@ -1223,6 +1265,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)
>@@ -1324,6 +1368,7 @@ static void *multifd_recv_thread(void *opaque)
>     trace_multifd_recv_thread_start(p->id);
>     rcu_register_thread();
> 
>+    multifd_recv_state->ops = &multifd_none_ops;

same as here.

>     while (true) {
>         uint32_t used;
>         uint32_t flags;
>@@ -1353,8 +1398,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;
>             }
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 7/8] multifd: Add zlib compression support

[Wei Yang]

On Wed, May 15, 2019 at 02:15:43PM +0200, Juan Quintela wrote:
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>---
> hw/core/qdev-properties.c |  2 +-
> migration/migration.c     |  9 ++++++++
> migration/migration.h     |  1 +
> migration/ram.c           | 47 +++++++++++++++++++++++++++++++++++++++
> qapi/migration.json       |  2 +-
> tests/migration-test.c    |  6 +++++
> 6 files changed, 65 insertions(+), 2 deletions(-)
>
>diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
>index ebeeb5c88d..e40aa806e2 100644
>--- a/hw/core/qdev-properties.c
>+++ b/hw/core/qdev-properties.c
>@@ -651,7 +651,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/migration.c b/migration/migration.c
>index d6f8ef342a..69d85cbe5e 100644
>--- a/migration/migration.c
>+++ b/migration/migration.c
>@@ -2141,6 +2141,15 @@ bool migrate_use_multifd(void)
>     return s->enabled_capabilities[MIGRATION_CAPABILITY_MULTIFD];
> }
> 
>+bool migrate_use_multifd_zlib(void)
>+{
>+    MigrationState *s;
>+
>+    s = migrate_get_current();
>+
>+    return s->parameters.multifd_compress == MULTIFD_COMPRESS_ZLIB;
>+}
>+
> bool migrate_pause_before_switchover(void)
> {
>     MigrationState *s;
>diff --git a/migration/migration.h b/migration/migration.h
>index 438f17edad..fc4fb841d4 100644
>--- a/migration/migration.h
>+++ b/migration/migration.h
>@@ -269,6 +269,7 @@ bool migrate_ignore_shared(void);
> 
> bool migrate_auto_converge(void);
> bool migrate_use_multifd(void);
>+bool migrate_use_multifd_zlib(void);
> bool migrate_pause_before_switchover(void);
> int migrate_multifd_channels(void);
> 
>diff --git a/migration/ram.c b/migration/ram.c
>index 6679e4f213..fdb5bf07a5 100644
>--- a/migration/ram.c
>+++ b/migration/ram.c
>@@ -582,6 +582,7 @@ exit:
> #define MULTIFD_VERSION 1
> 
> #define MULTIFD_FLAG_SYNC (1 << 0)
>+#define MULTIFD_FLAG_ZLIB (1 << 1)
> 

If no one use this in this patch, prefer to put it where it will be used.

> /* This value needs to be a multiple of qemu_target_page_size() */
> #define MULTIFD_PACKET_SIZE (512 * 1024)
>@@ -663,6 +664,12 @@ typedef struct {
>     uint64_t num_pages;
>     /* syncs main thread and channels */
>     QemuSemaphore sem_sync;
>+    /* stream for compression */
>+    z_stream zs;
>+    /* compressed buffer */
>+    uint8_t *zbuff;
>+    /* size of compressed buffer */
>+    uint32_t zbuff_len;
> }  MultiFDSendParams;
> 
> typedef struct {
>@@ -698,6 +705,12 @@ typedef struct {
>     uint64_t num_pages;
>     /* syncs main thread and channels */
>     QemuSemaphore sem_sync;
>+    /* stream for compression */
>+    z_stream zs;
>+    /* compressed buffer */
>+    uint8_t *zbuff;
>+    /* size of compressed buffer */
>+    uint32_t zbuff_len;
> } MultiFDRecvParams;
> 
> typedef struct {
>@@ -1071,6 +1084,9 @@ void multifd_save_cleanup(void)
>         p->packet_len = 0;
>         g_free(p->packet);
>         p->packet = NULL;
>+        deflateEnd(&p->zs);
>+        g_free(p->zbuff);
>+        p->zbuff = NULL;
>     }
>     qemu_sem_destroy(&multifd_send_state->channels_ready);
>     qemu_sem_destroy(&multifd_send_state->sem_sync);
>@@ -1240,6 +1256,7 @@ int multifd_save_setup(void)
> 
>     for (i = 0; i < thread_count; i++) {
>         MultiFDSendParams *p = &multifd_send_state->params[i];
>+        z_stream *zs = &p->zs;
> 
>         qemu_mutex_init(&p->mutex);
>         qemu_sem_init(&p->sem, 0);
>@@ -1253,6 +1270,17 @@ int multifd_save_setup(void)
>         p->packet = g_malloc0(p->packet_len);
>         p->name = g_strdup_printf("multifdsend_%d", i);
>         socket_send_channel_create(multifd_new_send_channel_async, p);
>+        zs->zalloc = Z_NULL;
>+        zs->zfree = Z_NULL;
>+        zs->opaque = Z_NULL;

Since zlib is not default option, is it better to setup these when zlib is
set?

>+        if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
>+            printf("deflate init failed\n");
>+            return -1;
>+        }
>+        /* We will never have more than page_count pages */
>+        p->zbuff_len = page_count * qemu_target_page_size();
>+        p->zbuff_len *= 2;
>+        p->zbuff = g_malloc0(p->zbuff_len);
>     }
>     return 0;
> }
>@@ -1322,6 +1350,9 @@ int multifd_load_cleanup(Error **errp)
>         p->packet_len = 0;
>         g_free(p->packet);
>         p->packet = NULL;
>+        inflateEnd(&p->zs);
>+        g_free(p->zbuff);
>+        p->zbuff = NULL;
>     }
>     qemu_sem_destroy(&multifd_recv_state->sem_sync);
>     g_free(multifd_recv_state->params);
>@@ -1440,6 +1471,7 @@ int multifd_load_setup(void)
> 
>     for (i = 0; i < thread_count; i++) {
>         MultiFDRecvParams *p = &multifd_recv_state->params[i];
>+        z_stream *zs = &p->zs;
> 
>         qemu_mutex_init(&p->mutex);
>         qemu_sem_init(&p->sem_sync, 0);
>@@ -1449,6 +1481,21 @@ int multifd_load_setup(void)
>                       + sizeof(ram_addr_t) * page_count;
>         p->packet = g_malloc0(p->packet_len);
>         p->name = g_strdup_printf("multifdrecv_%d", i);
>+
>+        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) {
>+            printf("inflate init failed\n");
>+            return -1;
>+        }
>+        /* We will never have more than page_count pages */
>+        p->zbuff_len = page_count * qemu_target_page_size();
>+        /* We know compression "could" use more space */
>+        p->zbuff_len *= 2;
>+        p->zbuff = g_malloc0(p->zbuff_len);
>     }
>     return 0;
> }
>diff --git a/qapi/migration.json b/qapi/migration.json
>index 8ec1944b7a..e6c27fae06 100644
>--- a/qapi/migration.json
>+++ b/qapi/migration.json
>@@ -493,7 +493,7 @@
> #
> ##
> { 'enum': 'MultifdCompress',
>-  'data': [ 'none' ] }
>+  'data': [ 'none', 'zlib' ] }
> 
> ##
> # @MigrationParameter:
>diff --git a/tests/migration-test.c b/tests/migration-test.c
>index 8a1ccc2516..2dd4d4c5b4 100644
>--- a/tests/migration-test.c
>+++ b/tests/migration-test.c
>@@ -1119,6 +1119,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";
>@@ -1174,6 +1179,7 @@ int main(int argc, char **argv)
>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>     qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
>+    qtest_add_func("/migration/multifd/tcp/zlib", test_multifd_tcp_zlib);

Actually zlib is not enabled at this moment, the test here may not touch the
real functionality.

> 
>     ret = g_test_run();
> 
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 8/8] multifd: rest of zlib compression

[Wei Yang]

On Wed, May 15, 2019 at 02:15:44PM +0200, Juan Quintela wrote:
>This is still a work in progress, but get everything sent as expected
>and it is faster than the code that is already there.

Generally, I prefer to merge this one with previous one.

>
>Signed-off-by: Juan Quintela <quintela@redhat.com>
>---
> migration/ram.c | 106 +++++++++++++++++++++++++++++++++++++++++++++++-
> 1 file changed, 104 insertions(+), 2 deletions(-)
>
>diff --git a/migration/ram.c b/migration/ram.c
>index fdb5bf07a5..efbb253c1a 100644
>--- a/migration/ram.c
>+++ b/migration/ram.c
>@@ -747,6 +747,100 @@ MultifdMethods multifd_none_ops = {
>     .recv_pages = none_recv_pages
> };
> 
>+/* Multifd zlib compression */
>+
>+static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used)
>+{
>+    struct iovec *iov = p->pages->iov;
>+    z_stream *zs = &p->zs;
>+    uint32_t out_size = 0;
>+    int ret;
>+    int i;
>+
>+    for (i = 0; i < used; i++) {
>+        uint32_t available = p->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 = p->zbuff + out_size;
>+
>+        ret = deflate(zs, flush);
>+        if (ret != Z_OK) {
>+            printf("problem with deflate? %d\n", ret);
>+            qemu_mutex_unlock(&p->mutex);
>+            return -1;
>+        }
>+        out_size += available - zs->avail_out;
>+    }
>+    p->next_packet_size = out_size;
>+
>+    return 0;
>+}
>+
>+static int zlib_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
>+{
>+    return qio_channel_write_all(p->c, (void *)p->zbuff, p->next_packet_size,
>+                                 perr);
>+}
>+
>+static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
>+{
>+    uint32_t in_size = p->next_packet_size;
>+    uint32_t out_size = 0;
>+    uint32_t expected_size = used * qemu_target_page_size();
>+    z_stream *zs = &p->zs;
>+    int ret;
>+    int i;
>+
>+    ret = qio_channel_read_all(p->c, (void *)p->zbuff, in_size, perr);
>+
>+    if (ret != 0) {
>+        return ret;
>+    }
>+
>+    zs->avail_in = in_size;
>+    zs->next_in = p->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) {
>+            printf("%d: problem with inflate? %d\n", p->id, ret);
>+            qemu_mutex_unlock(&p->mutex);
>+            return ret;
>+        }
>+        out_size += iov->iov_len;
>+    }
>+    if (out_size != expected_size) {
>+        printf("out size %d expected size %d\n",
>+               out_size, expected_size);
>+        return -1;
>+    }
>+    return 0;
>+}
>+
>+MultifdMethods multifd_zlib_ops = {
>+    .send_prepare = zlib_send_prepare,
>+    .send_write = zlib_send_write,
>+    .recv_pages = zlib_recv_pages
>+};
>+
> static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
> {
>     MultiFDInit_t msg;
>@@ -1145,7 +1239,11 @@ static void *multifd_send_thread(void *opaque)
>     /* initial packet */
>     p->num_packets = 1;
> 
>-    multifd_send_state->ops = &multifd_none_ops;
>+    if (migrate_use_multifd_zlib()) {
>+        multifd_send_state->ops = &multifd_zlib_ops;
>+    } else {
>+        multifd_send_state->ops = &multifd_none_ops;
>+    }

Again, to manipulate a global variable in each thread is not a good idea.

This would be better to use an array to assign ops instead of *if*. In case
you would have several compress methods, the code would be difficult to read.

> 
>     while (true) {
>         qemu_sem_wait(&p->sem);
>@@ -1399,7 +1497,11 @@ static void *multifd_recv_thread(void *opaque)
>     trace_multifd_recv_thread_start(p->id);
>     rcu_register_thread();
> 
>-    multifd_recv_state->ops = &multifd_none_ops;
>+    if (migrate_use_multifd_zlib()) {
>+        multifd_recv_state->ops = &multifd_zlib_ops;
>+    } else {
>+        multifd_recv_state->ops = &multifd_none_ops;
>+    }
>     while (true) {
>         uint32_t used;
>         uint32_t flags;
>-- 
>2.21.0
>

-- 
Wei Yang
Help you, Help me

Re: [Qemu-devel] [PATCH v3 6/8] migration: Make none operations into its own structure

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> It will be used later.

'none' is confusing - I think this is no-compression specifically -
right?
I'd be happy with something abbreviated like 'nocomp'

> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c | 54 ++++++++++++++++++++++++++++++++++++++++++++-----
>  1 file changed, 49 insertions(+), 5 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index 1ca9ba77b6..6679e4f213 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -700,6 +700,40 @@ typedef struct {
>      QemuSemaphore sem_sync;
>  } MultiFDRecvParams;
>  
> +typedef struct {
> +    /* Prepare the send packet */
> +    int (*send_prepare)(MultiFDSendParams *p, uint32_t used);
> +    /* Write the send packet */
> +    int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **perr);
> +    /* Read all pages */
> +    int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **perr);
> +} MultifdMethods;
> +
> +/* Multifd without compression */
> +
> +static int none_send_prepare(MultiFDSendParams *p, uint32_t used)
> +{
> +    p->next_packet_size = used * qemu_target_page_size();
> +    return 0;
> +}
> +
> +static int none_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
> +{
> +    return qio_channel_writev_all(p->c, p->pages->iov, used, perr);
> +}
> +
> +static int none_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
> +{
> +    return qio_channel_readv_all(p->c, p->pages->iov, used, perr);
> +
> +}
> +
> +MultifdMethods multifd_none_ops = {
> +    .send_prepare = none_send_prepare,
> +    .send_write = none_send_write,
> +    .recv_pages = none_recv_pages
> +};
> +
>  static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
>  {
>      MultiFDInit_t msg;
> @@ -904,6 +938,8 @@ struct {
>      uint64_t packet_num;
>      /* send channels ready */
>      QemuSemaphore channels_ready;
> +    /* multifd ops */
> +    MultifdMethods *ops;
>  } *multifd_send_state;
>  
>  /*
> @@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
>      /* initial packet */
>      p->num_packets = 1;
>  
> +    multifd_send_state->ops = &multifd_none_ops;
> +

I agree with Wei Yang that is a bad idea; that should be done once
before the first thread is started.

Dave

>      while (true) {
>          qemu_sem_wait(&p->sem);
>          qemu_mutex_lock(&p->mutex);
> @@ -1102,7 +1140,12 @@ static void *multifd_send_thread(void *opaque)
>              uint64_t packet_num = p->packet_num;
>              uint32_t flags = p->flags;
>  
> -            p->next_packet_size = used * qemu_target_page_size();
> +            if (used) {
> +                ret = multifd_send_state->ops->send_prepare(p, used);
> +                if (ret != 0) {
> +                    break;
> +                }
> +            }
>              multifd_send_fill_packet(p);
>              p->flags = 0;
>              p->num_packets++;
> @@ -1120,8 +1163,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;
>                  }
> @@ -1223,6 +1265,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)
> @@ -1324,6 +1368,7 @@ static void *multifd_recv_thread(void *opaque)
>      trace_multifd_recv_thread_start(p->id);
>      rcu_register_thread();
>  
> +    multifd_recv_state->ops = &multifd_none_ops;
>      while (true) {
>          uint32_t used;
>          uint32_t flags;
> @@ -1353,8 +1398,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;
>              }
> -- 
> 2.21.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v3 7/8] multifd: Add zlib compression support

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  hw/core/qdev-properties.c |  2 +-
>  migration/migration.c     |  9 ++++++++
>  migration/migration.h     |  1 +
>  migration/ram.c           | 47 +++++++++++++++++++++++++++++++++++++++
>  qapi/migration.json       |  2 +-
>  tests/migration-test.c    |  6 +++++
>  6 files changed, 65 insertions(+), 2 deletions(-)
> 
> diff --git a/hw/core/qdev-properties.c b/hw/core/qdev-properties.c
> index ebeeb5c88d..e40aa806e2 100644
> --- a/hw/core/qdev-properties.c
> +++ b/hw/core/qdev-properties.c
> @@ -651,7 +651,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/migration.c b/migration/migration.c
> index d6f8ef342a..69d85cbe5e 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -2141,6 +2141,15 @@ bool migrate_use_multifd(void)
>      return s->enabled_capabilities[MIGRATION_CAPABILITY_MULTIFD];
>  }
>  
> +bool migrate_use_multifd_zlib(void)
> +{
> +    MigrationState *s;
> +
> +    s = migrate_get_current();
> +
> +    return s->parameters.multifd_compress == MULTIFD_COMPRESS_ZLIB;
> +}
> +
>  bool migrate_pause_before_switchover(void)
>  {
>      MigrationState *s;
> diff --git a/migration/migration.h b/migration/migration.h
> index 438f17edad..fc4fb841d4 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -269,6 +269,7 @@ bool migrate_ignore_shared(void);
>  
>  bool migrate_auto_converge(void);
>  bool migrate_use_multifd(void);
> +bool migrate_use_multifd_zlib(void);
>  bool migrate_pause_before_switchover(void);
>  int migrate_multifd_channels(void);
>  
> diff --git a/migration/ram.c b/migration/ram.c
> index 6679e4f213..fdb5bf07a5 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -582,6 +582,7 @@ exit:
>  #define MULTIFD_VERSION 1
>  
>  #define MULTIFD_FLAG_SYNC (1 << 0)
> +#define MULTIFD_FLAG_ZLIB (1 << 1)
>  
>  /* This value needs to be a multiple of qemu_target_page_size() */
>  #define MULTIFD_PACKET_SIZE (512 * 1024)
> @@ -663,6 +664,12 @@ typedef struct {
>      uint64_t num_pages;
>      /* syncs main thread and channels */
>      QemuSemaphore sem_sync;
> +    /* stream for compression */
> +    z_stream zs;
> +    /* compressed buffer */
> +    uint8_t *zbuff;
> +    /* size of compressed buffer */
> +    uint32_t zbuff_len;

Does this set need to be in a union or something so that you select
them for different compression types?

>  }  MultiFDSendParams;
>  
>  typedef struct {
> @@ -698,6 +705,12 @@ typedef struct {
>      uint64_t num_pages;
>      /* syncs main thread and channels */
>      QemuSemaphore sem_sync;
> +    /* stream for compression */

de-compression?

> +    z_stream zs;
> +    /* compressed buffer */
> +    uint8_t *zbuff;
> +    /* size of compressed buffer */
> +    uint32_t zbuff_len;
>  } MultiFDRecvParams;
>  
>  typedef struct {
> @@ -1071,6 +1084,9 @@ void multifd_save_cleanup(void)
>          p->packet_len = 0;
>          g_free(p->packet);
>          p->packet = NULL;
> +        deflateEnd(&p->zs);
> +        g_free(p->zbuff);
> +        p->zbuff = NULL;
>      }
>      qemu_sem_destroy(&multifd_send_state->channels_ready);
>      qemu_sem_destroy(&multifd_send_state->sem_sync);
> @@ -1240,6 +1256,7 @@ int multifd_save_setup(void)
>  
>      for (i = 0; i < thread_count; i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
> +        z_stream *zs = &p->zs;
>  
>          qemu_mutex_init(&p->mutex);
>          qemu_sem_init(&p->sem, 0);
> @@ -1253,6 +1270,17 @@ int multifd_save_setup(void)
>          p->packet = g_malloc0(p->packet_len);
>          p->name = g_strdup_printf("multifdsend_%d", i);
>          socket_send_channel_create(multifd_new_send_channel_async, p);
> +        zs->zalloc = Z_NULL;
> +        zs->zfree = Z_NULL;
> +        zs->opaque = Z_NULL;
> +        if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
> +            printf("deflate init failed\n");
> +            return -1;
> +        }
> +        /* We will never have more than page_count pages */
> +        p->zbuff_len = page_count * qemu_target_page_size();
> +        p->zbuff_len *= 2;

Should the ops gain a 'save_init' and 'load_init' so that you can
only do this lot if the compression is enabled?

> +        p->zbuff = g_malloc0(p->zbuff_len);

I'd prefer g_try_malloc and do failure given it's not a tiny buffer.

>      }
>      return 0;
>  }
> @@ -1322,6 +1350,9 @@ int multifd_load_cleanup(Error **errp)
>          p->packet_len = 0;
>          g_free(p->packet);
>          p->packet = NULL;
> +        inflateEnd(&p->zs);
> +        g_free(p->zbuff);
> +        p->zbuff = NULL;
>      }
>      qemu_sem_destroy(&multifd_recv_state->sem_sync);
>      g_free(multifd_recv_state->params);
> @@ -1440,6 +1471,7 @@ int multifd_load_setup(void)
>  
>      for (i = 0; i < thread_count; i++) {
>          MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +        z_stream *zs = &p->zs;
>  
>          qemu_mutex_init(&p->mutex);
>          qemu_sem_init(&p->sem_sync, 0);
> @@ -1449,6 +1481,21 @@ int multifd_load_setup(void)
>                        + sizeof(ram_addr_t) * page_count;
>          p->packet = g_malloc0(p->packet_len);
>          p->name = g_strdup_printf("multifdrecv_%d", i);
> +
> +        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) {
> +            printf("inflate init failed\n");
> +            return -1;
> +        }
> +        /* We will never have more than page_count pages */
> +        p->zbuff_len = page_count * qemu_target_page_size();
> +        /* We know compression "could" use more space */
> +        p->zbuff_len *= 2;
> +        p->zbuff = g_malloc0(p->zbuff_len);
>      }
>      return 0;
>  }
> diff --git a/qapi/migration.json b/qapi/migration.json
> index 8ec1944b7a..e6c27fae06 100644
> --- a/qapi/migration.json
> +++ b/qapi/migration.json
> @@ -493,7 +493,7 @@
>  #
>  ##
>  { 'enum': 'MultifdCompress',
> -  'data': [ 'none' ] }
> +  'data': [ 'none', 'zlib' ] }
>  
>  ##
>  # @MigrationParameter:
> diff --git a/tests/migration-test.c b/tests/migration-test.c
> index 8a1ccc2516..2dd4d4c5b4 100644
> --- a/tests/migration-test.c
> +++ b/tests/migration-test.c
> @@ -1119,6 +1119,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";
> @@ -1174,6 +1179,7 @@ int main(int argc, char **argv)
>      /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>      qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>      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.21.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v3 8/8] multifd: rest of zlib compression

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> This is still a work in progress, but get everything sent as expected
> and it is faster than the code that is already there.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c | 106 +++++++++++++++++++++++++++++++++++++++++++++++-
>  1 file changed, 104 insertions(+), 2 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index fdb5bf07a5..efbb253c1a 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -747,6 +747,100 @@ MultifdMethods multifd_none_ops = {
>      .recv_pages = none_recv_pages
>  };
>  
> +/* Multifd zlib compression */
> +

Comment the return value?

> +static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used)
> +{
> +    struct iovec *iov = p->pages->iov;
> +    z_stream *zs = &p->zs;
> +    uint32_t out_size = 0;
> +    int ret;
> +    int i;

uint32_t to match 'used' ?

> +    for (i = 0; i < used; i++) {
> +        uint32_t available = p->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 = p->zbuff + out_size;
> +
> +        ret = deflate(zs, flush);
> +        if (ret != Z_OK) {
> +            printf("problem with deflate? %d\n", ret);

If it's an error it should probably be at least an fprintf(stderr or
err_ something.

Should this also check that the avail_in/next_in has consumed the whole
of the input?

> +            qemu_mutex_unlock(&p->mutex);

Can you explain and/or comment whyit's unlocked here in the error path?

> +            return -1;
> +        }
> +        out_size += available - zs->avail_out;
> +    }
> +    p->next_packet_size = out_size;

Some traces_ wouldn't hurt.

> +    return 0;
> +}
> +
> +static int zlib_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
> +{
> +    return qio_channel_write_all(p->c, (void *)p->zbuff, p->next_packet_size,
> +                                 perr);
> +}
> +
> +static int zlib_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
> +{
> +    uint32_t in_size = p->next_packet_size;
> +    uint32_t out_size = 0;
> +    uint32_t expected_size = used * qemu_target_page_size();
> +    z_stream *zs = &p->zs;
> +    int ret;
> +    int i;
> +
> +    ret = qio_channel_read_all(p->c, (void *)p->zbuff, in_size, perr);
> +
> +    if (ret != 0) {
> +        return ret;
> +    }
> +
> +    zs->avail_in = in_size;
> +    zs->next_in = p->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) {
> +            printf("%d: problem with inflate? %d\n", p->id, ret);
> +            qemu_mutex_unlock(&p->mutex);
> +            return ret;
> +        }
> +        out_size += iov->iov_len;
> +    }
> +    if (out_size != expected_size) {
> +        printf("out size %d expected size %d\n",
> +               out_size, expected_size);
> +        return -1;
> +    }
> +    return 0;
> +}
> +
> +MultifdMethods multifd_zlib_ops = {
> +    .send_prepare = zlib_send_prepare,
> +    .send_write = zlib_send_write,
> +    .recv_pages = zlib_recv_pages
> +};
> +
>  static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
>  {
>      MultiFDInit_t msg;
> @@ -1145,7 +1239,11 @@ static void *multifd_send_thread(void *opaque)
>      /* initial packet */
>      p->num_packets = 1;
>  
> -    multifd_send_state->ops = &multifd_none_ops;
> +    if (migrate_use_multifd_zlib()) {
> +        multifd_send_state->ops = &multifd_zlib_ops;
> +    } else {
> +        multifd_send_state->ops = &multifd_none_ops;
> +    }
>  
>      while (true) {
>          qemu_sem_wait(&p->sem);
> @@ -1399,7 +1497,11 @@ static void *multifd_recv_thread(void *opaque)
>      trace_multifd_recv_thread_start(p->id);
>      rcu_register_thread();
>  
> -    multifd_recv_state->ops = &multifd_none_ops;
> +    if (migrate_use_multifd_zlib()) {
> +        multifd_recv_state->ops = &multifd_zlib_ops;
> +    } else {
> +        multifd_recv_state->ops = &multifd_none_ops;
> +    }
>      while (true) {
>          uint32_t used;
>          uint32_t flags;
> -- 
> 2.21.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v3 3/8] tests: Add migration multifd test

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:39PM +0200, Juan Quintela wrote:
>>We set multifd-channels.
>>
>>Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>>Reviewed-by: Thomas Huth <thuth@redhat.com>
>>Signed-off-by: Juan Quintela <quintela@redhat.com>
>>---
>> tests/migration-test.c | 48 ++++++++++++++++++++++++++++++++++++++++++
>> 1 file changed, 48 insertions(+)
>>
>>diff --git a/tests/migration-test.c b/tests/migration-test.c
>>index 0b25aa3d6c..ff480e0682 100644
>>--- a/tests/migration-test.c
>>+++ b/tests/migration-test.c
>>@@ -1028,6 +1028,53 @@ static void test_precopy_tcp(void)
>>     g_free(uri);
>> }
>> 
>>+static void test_multifd_tcp(void)
>>+{
>>+    char *uri;
>>+    QTestState *from, *to;
>>+
>>+    if (test_migrate_start(&from, &to, "tcp:127.0.0.1:0", false, false)) {
>>+        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", 2);
>>+    migrate_set_parameter_int(to, "multifd-channels", 2);
>>+
>>+    migrate_set_capability(from, "multifd", "true");
>>+    migrate_set_capability(to, "multifd", "true");
>
> Hi, Juan
>
> In case to test multifd, what's the minimum configuration? Set multifd
> capability at both side is fine?

It is required.  For older configuration, we are a bit more lax about
that requirement.  For anything new, we want it to be set in both sides.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 3/8] tests: Add migration multifd test

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:39PM +0200, Juan Quintela wrote:
>>We set multifd-channels.
>>
>>+    wait_for_serial("dest_serial");
>>+    wait_for_migration_complete(from);
>>+
>>+    test_migrate_end(from, to, true);
>
> Miss free uri here?
>
>     g_free(uri);

You are right.
Fixed.

>
> And
>
> Tested-by: Wei Yang <richardw.yang@linux.intel.com>

Thanks.

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Juan Quintela]

Markus Armbruster <armbru@redhat.com> wrote:
> Juan Quintela <quintela@redhat.com> writes:
>> +    case MIGRATION_PARAMETER_MULTIFD_COMPRESS:
>> +        p->has_multifd_compress = true;
>> +        visit_type_MultifdCompress(v, param, &compress_type, &err);
>> +        if (err) {
>> +            break;
>> +        }
>> +        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {
>> +            error_setg(&err, "Invalid multifd_compress option %s", valuestr);
>> +            break;
>> +        }
>
> This should never happen.  If you want to check anyway, make it an
> assertion.
>
> Just in case you don't believe me, or are curious:
>
> visit_type_MultifdCompress() wraps around visit_type_enum(), passing it
> &MultifdCompress_lookup.
>
> Since @v is an input visitor, visit_type_enum() wraps around
> input_type_enum().
>
> input_type_enum() computes the value to store in @compress_type with
> qapi_enum_parse().
>
> To get here, visit_type_MultifdCompress() must have succeeded,
> i.e. visit_type_enum(), input_type_enum() and qapi_enum_parse() all
> succeded.
>
> On success, qapi_enum_parse() returns one of the values in
> MultifdCompress_lookup, i.e. a member of enum MultifdCompress other than
> MULTIFD_COMPRESS__MAX.

Fixed, thanks.

>> @@ -3353,6 +3362,9 @@ void migration_global_dump(Monitor *mon)
>>  #define DEFINE_PROP_MIG_CAP(name, x)             \
>>      DEFINE_PROP_BOOL(name, MigrationState, enabled_capabilities[x], false)
>>  
>> +#define DEFINE_PROP_MULTIFD_COMPRESS(_n, _s, _f, _d) \
>> +    DEFINE_PROP_SIGNED(_n, _s, _f, _d, qdev_prop_multifd_compress, MultifdCompress)
>> +
>
> Did you forget to move this?

It appears that yes.  I tried to move the other part out of
qdev-properties, failed.  And then forgot to move this bit.

Thanks.

Later, Juan.

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>>Signed-off-by: Juan Quintela <quintela@redhat.com>
>>diff --git a/tests/migration-test.c b/tests/migration-test.c
>>index 65d5e256a7..8a1ccc2516 100644
>>--- a/tests/migration-test.c
>>+++ b/tests/migration-test.c
>
> Well, may I suggest to split the test into another one?

In next patch, I get another multifd method.  Think of this patch as a
"rename", more than something different.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Juan Quintela]

Wei Yang <richard.weiyang@gmail.com> wrote:
> On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>>Signed-off-by: Juan Quintela <quintela@redhat.com>
>>@@ -1821,6 +1826,18 @@ 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;
>>+        }
>>+        if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {
>
> A warning during build:
>
> hmp.c:1835:27: warning: comparison of unsigned enum expression < 0 is
> always false [-Wtautological-compare]
>         if (compress_type < 0 || compress_type >= MULTIFD_COMPRESS__MAX) {

Fixed, see Markus reason for dropping this bit.

/me wonders why this didn't failed my compilation ....

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 5/8] migration: Add multifd-compress parameter

[Juan Quintela]

Wei Yang <richard.weiyang@gmail.com> wrote:
> On Wed, May 15, 2019 at 02:15:41PM +0200, Juan Quintela wrote:
>>diff --git a/qapi/migration.json b/qapi/migration.json
>>index 9cfbaf8c6c..8ec1944b7a 100644
>>--- a/qapi/migration.json
>>+++ b/qapi/migration.json
>>@@ -482,6 +482,19 @@
>> # TODO either fuse back into MigrationParameters, or make
>>@@ -707,7 +726,8 @@
>>             '*multifd-channels': 'int',
>>             '*xbzrle-cache-size': 'size',
>>             '*max-postcopy-bandwidth': 'size',
>>-	    '*max-cpu-throttle': 'int' } }
>>+	    '*max-cpu-throttle': 'int',
>
> A tab at the beginning, it would be better to fix this :-)

The wonders of magit+emacs, I didn't saw this one.

Fixed.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 6/8] migration: Make none operations into its own structure

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:42PM +0200, Juan Quintela wrote:
>>+
>>+MultifdMethods multifd_none_ops = {
>>+    .send_prepare = none_send_prepare,
>>+    .send_write = none_send_write,
>>+    .recv_pages = none_recv_pages
>>+};
>>+
>> static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
>> {
>>     MultiFDInit_t msg;
>>@@ -904,6 +938,8 @@ struct {
>>     uint64_t packet_num;
>>     /* send channels ready */
>>     QemuSemaphore channels_ready;
>>+    /* multifd ops */
>>+    MultifdMethods *ops;
>> } *multifd_send_state;
>> 
>> /*
>>@@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
>>     /* initial packet */
>>     p->num_packets = 1;
>> 
>>+    multifd_send_state->ops = &multifd_none_ops;
>>+
>
> I am afraid it is not a good practice to assign ops when each thread starts
> work.

Agreed.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 6/8] migration: Make none 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.
>
> 'none' is confusing - I think this is no-compression specifically -
> right?
> I'd be happy with something abbreviated like 'nocomp'

Got into nocomp.
>> @@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
>>      /* initial packet */
>>      p->num_packets = 1;
>>  
>> +    multifd_send_state->ops = &multifd_none_ops;
>> +
>
> I agree with Wei Yang that is a bad idea; that should be done once
> before the first thread is started.

Also fixed.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 0/8] WIP: Multifd compression support

[Eric Blake]

[-- Attachment #1.1: Type: text/plain, Size: 904 bytes --]

On 5/20/19 1:35 AM, Wei Yang wrote:
> On Wed, May 15, 2019 at 02:15:36PM +0200, Juan Quintela wrote:
>> v3:
>> - improve the code
>> - address David and Markus comments
>> - make compression code into methods
>>  so we can add any other method ading just three functions
>>
>> Please review, as far as I know everything is ok now.
>>
>> Todo: Add zstd support
> 
> Confusion here. It is zstd or sztd?
> 
> BTW, I am not sure what it is :-)
> 
>>
>> v2:
>> - improve the code left and right
>> - Split better the zlib code
>> - rename everything to v4.1
>> - Add tests for multifd-compress zlib
>> - Parameter is now an enum (soon will see sztd)
>                                            ^^^

zstd is the name of the new compression algorithm.

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3226
Virtualization:  qemu.org | libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [Qemu-devel] [PATCH v3 7/8] multifd: Add zlib compression support

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  hw/core/qdev-properties.c |  2 +-
>>  migration/migration.c     |  9 ++++++++
>>  migration/migration.h     |  1 +
>>  migration/ram.c           | 47 +++++++++++++++++++++++++++++++++++++++
>>  qapi/migration.json       |  2 +-
>>  tests/migration-test.c    |  6 +++++
>>  6 files changed, 65 insertions(+), 2 deletions(-)
>> 

>>  /* This value needs to be a multiple of qemu_target_page_size() */
>>  #define MULTIFD_PACKET_SIZE (512 * 1024)
>> @@ -663,6 +664,12 @@ typedef struct {
>>      uint64_t num_pages;
>>      /* syncs main thread and channels */
>>      QemuSemaphore sem_sync;
>> +    /* stream for compression */
>> +    z_stream zs;
>> +    /* compressed buffer */
>> +    uint8_t *zbuff;
>> +    /* size of compressed buffer */
>> +    uint32_t zbuff_len;
>
> Does this set need to be in a union or something so that you select
> them for different compression types?

Yeap.  Done.

>
>>  }  MultiFDSendParams;
>>  
>>  typedef struct {
>> @@ -698,6 +705,12 @@ typedef struct {
>>      uint64_t num_pages;
>>      /* syncs main thread and channels */
>>      QemuSemaphore sem_sync;
>> +    /* stream for compression */
>
> de-compression?

Changed.  I think that "compression methods" mean both, but who I am to
discuss with a native speaker O:-)

>>          socket_send_channel_create(multifd_new_send_channel_async, p);
>> +        zs->zalloc = Z_NULL;
>> +        zs->zfree = Z_NULL;
>> +        zs->opaque = Z_NULL;
>> +        if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
>> +            printf("deflate init failed\n");
>> +            return -1;
>> +        }
>> +        /* We will never have more than page_count pages */
>> +        p->zbuff_len = page_count * qemu_target_page_size();
>> +        p->zbuff_len *= 2;
>
> Should the ops gain a 'save_init' and 'load_init' so that you can
> only do this lot if the compression is enabled?

send_setup()/send_cleanup()
recv_setup()/recv_cleanup()

I have tried to be consistent ....

>
>> +        p->zbuff = g_malloc0(p->zbuff_len);
>
> I'd prefer g_try_malloc and do failure given it's not a tiny buffer.

I can change, no problem there.
Changing prototypes to get an Error *.


Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 7/8] multifd: Add zlib compression support

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:43PM +0200, Juan Quintela wrote:

>> 
>> #define MULTIFD_FLAG_SYNC (1 << 0)
>>+#define MULTIFD_FLAG_ZLIB (1 << 1)
>> 
>
> If no one use this in this patch, prefer to put it where it will be used.

Oops, you are right, I have to use it.

>>@@ -1253,6 +1270,17 @@ int multifd_save_setup(void)
>>         p->packet = g_malloc0(p->packet_len);
>>         p->name = g_strdup_printf("multifdsend_%d", i);
>>         socket_send_channel_create(multifd_new_send_channel_async, p);
>>+        zs->zalloc = Z_NULL;
>>+        zs->zfree = Z_NULL;
>>+        zs->opaque = Z_NULL;
>
> Since zlib is not default option, is it better to setup these when zlib is
> set?

Moved to an opaque void *data pointer, thanks.

>>+        if (deflateInit(zs, migrate_compress_level()) != Z_OK) {
>>+            printf("deflate init failed\n");
>>+            return -1;
>>+        }
>>+        /* We will never have more than page_count pages */
>>+        p->zbuff_len = page_count * qemu_target_page_size();
>>+        p->zbuff_len *= 2;
>>+        p->zbuff = g_malloc0(p->zbuff_len);
>>     }
>>     return 0;
>> }
>>@@ -1322,6 +1350,9 @@ int multifd_load_cleanup(Error **errp)
>>         p->packet_len = 0;
>>         g_free(p->packet);
>>         p->packet = NULL;
>>+        inflateEnd(&p->zs);
>>+        g_free(p->zbuff);
>>+        p->zbuff = NULL;
>>     }
>>     qemu_sem_destroy(&multifd_recv_state->sem_sync);
>>     g_free(multifd_recv_state->params);
>>@@ -1440,6 +1471,7 @@ int multifd_load_setup(void)
>> 
>>     for (i = 0; i < thread_count; i++) {
>>         MultiFDRecvParams *p = &multifd_recv_state->params[i];
>>+        z_stream *zs = &p->zs;
>> 
>>         qemu_mutex_init(&p->mutex);
>>         qemu_sem_init(&p->sem_sync, 0);
>>@@ -1449,6 +1481,21 @@ int multifd_load_setup(void)
>>                       + sizeof(ram_addr_t) * page_count;
>>         p->packet = g_malloc0(p->packet_len);
>>         p->name = g_strdup_printf("multifdrecv_%d", i);
>>+
>>+        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) {
>>+            printf("inflate init failed\n");
>>+            return -1;
>>+        }
>>+        /* We will never have more than page_count pages */
>>+        p->zbuff_len = page_count * qemu_target_page_size();
>>+        /* We know compression "could" use more space */
>>+        p->zbuff_len *= 2;
>>+        p->zbuff = g_malloc0(p->zbuff_len);
>>     }
>>     return 0;
>> }
>>diff --git a/qapi/migration.json b/qapi/migration.json
>>index 8ec1944b7a..e6c27fae06 100644
>>--- a/qapi/migration.json
>>+++ b/qapi/migration.json
>>@@ -493,7 +493,7 @@
>> #
>> ##
>> { 'enum': 'MultifdCompress',
>>-  'data': [ 'none' ] }
>>+  'data': [ 'none', 'zlib' ] }
>> 
>> ##
>> # @MigrationParameter:
>>diff --git a/tests/migration-test.c b/tests/migration-test.c
>>index 8a1ccc2516..2dd4d4c5b4 100644
>>--- a/tests/migration-test.c
>>+++ b/tests/migration-test.c
>>@@ -1119,6 +1119,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";
>>@@ -1174,6 +1179,7 @@ int main(int argc, char **argv)
>>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>>     qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
>>+    qtest_add_func("/migration/multifd/tcp/zlib", test_multifd_tcp_zlib);
>
> Actually zlib is not enabled at this moment, the test here may not touch the
> real functionality.

It is, see what the

   test_multifd_tcp("zlib");

line does for:


>>     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";
>>@@ -1174,6 +1179,7 @@ int main(int argc, char **argv)
>>     /* qtest_add_func("/migration/ignore_shared", test_ignore_shared); */
>>     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
>>     qtest_add_func("/migration/multifd/tcp/none", test_multifd_tcp_none);
>>+    qtest_add_func("/migration/multifd/tcp/zlib", test_multifd_tcp_zlib);
>
> Actually zlib is not enabled at this moment, the test here may not touch the
> real functionality.

It is, see what the

   test_multifd_tcp("zlib");

line does for:

static void test_multifd_tcp(const char *method)
{
    ....
    migrate_set_parameter_str(from, "multifd-compress", method);
    migrate_set_parameter_str(to, "multifd-compress", method);

    ...
}

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 8/8] multifd: rest of zlib compression

[Juan Quintela]

Wei Yang <richardw.yang@linux.intel.com> wrote:
> On Wed, May 15, 2019 at 02:15:44PM +0200, Juan Quintela wrote:
>>This is still a work in progress, but get everything sent as expected
>>and it is faster than the code that is already there.
>
> Generally, I prefer to merge this one with previous one.

Done, sir O:-)

For the WIP part, it was easier to have the bits that didn't change and
the ones that I was working with.

>>@@ -1145,7 +1239,11 @@ static void *multifd_send_thread(void *opaque)
>>     /* initial packet */
>>     p->num_packets = 1;
>> 
>>-    multifd_send_state->ops = &multifd_none_ops;
>>+    if (migrate_use_multifd_zlib()) {
>>+        multifd_send_state->ops = &multifd_zlib_ops;
>>+    } else {
>>+        multifd_send_state->ops = &multifd_none_ops;
>>+    }
>
> Again, to manipulate a global variable in each thread is not a good idea.

Fixed.

> This would be better to use an array to assign ops instead of *if*. In case
> you would have several compress methods, the code would be difficult to read.

it is going to end:

   if (migrate_use_multifd_zlib()) {
       multifd_send_state->ops = &multifd_zlib_ops;
   if (migrate_use_multifd_zstd()) {
       multifd_send_state->ops = &multifd_zstd_ops;
   } else {
       multifd_send_state->ops = &multifd_none_ops;
   }

We can use:

multifd_send_state->ops = multifd_ops[migrate_multifd_method(void)];

About what is easier to read .....  it depends on taste.

Will change anyways.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 8/8] multifd: rest of zlib compression

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> This is still a work in progress, but get everything sent as expected
>> and it is faster than the code that is already there.
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/ram.c | 106 +++++++++++++++++++++++++++++++++++++++++++++++-
>>  1 file changed, 104 insertions(+), 2 deletions(-)
>> 
>> diff --git a/migration/ram.c b/migration/ram.c
>> index fdb5bf07a5..efbb253c1a 100644
>> --- a/migration/ram.c
>> +++ b/migration/ram.c
>> @@ -747,6 +747,100 @@ MultifdMethods multifd_none_ops = {
>>      .recv_pages = none_recv_pages
>>  };
>>  
>> +/* Multifd zlib compression */
>> +
>
> Comment the return value?

Once there, commented all the functions.

>> +static int zlib_send_prepare(MultiFDSendParams *p, uint32_t used)
>> +{
>> +    struct iovec *iov = p->pages->iov;
>> +    z_stream *zs = &p->zs;
>> +    uint32_t out_size = 0;
>> +    int ret;
>> +    int i;
>
> uint32_t to match 'used' ?

Done

>> +    for (i = 0; i < used; i++) {
>> +        uint32_t available = p->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 = p->zbuff + out_size;
>> +
>> +        ret = deflate(zs, flush);
>> +        if (ret != Z_OK) {
>> +            printf("problem with deflate? %d\n", ret);
>
> If it's an error it should probably be at least an fprintf(stderr or
> err_ something.

We don't have any error arround really, we need one. Searching for it.

> Should this also check that the avail_in/next_in has consumed the whole
> of the input?

I am not checking because _it_ is supposed to b doing it right.  We can
test it through, specially in reception.

>> +            qemu_mutex_unlock(&p->mutex);
>
> Can you explain and/or comment whyit's unlocked here in the error path?

Uh, oh ....

Leftover for when it was done inline inside the main function.
Removed.

>> +            return -1;
>> +        }
>> +        out_size += available - zs->avail_out;
>> +    }
>> +    p->next_packet_size = out_size;
>
> Some traces_ wouldn't hurt.

Humm, you are right here.

Thanks, Juan.

Re: [Qemu-devel] [PATCH v3 6/8] migration: Make none 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.
>
> 'none' is confusing - I think this is no-compression specifically -
> right?
> I'd be happy with something abbreviated like 'nocomp'

I don't care too much, I can change, but when you are setting the value
it gets:

micgration set-parameter compression none

That looks ok.

On the other hand, I can agree that I can call the functions nocomp.

Thanks, Juan.

>
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/ram.c | 54 ++++++++++++++++++++++++++++++++++++++++++++-----
>>  1 file changed, 49 insertions(+), 5 deletions(-)
>> 
>> diff --git a/migration/ram.c b/migration/ram.c
>> index 1ca9ba77b6..6679e4f213 100644
>> --- a/migration/ram.c
>> +++ b/migration/ram.c
>> @@ -700,6 +700,40 @@ typedef struct {
>>      QemuSemaphore sem_sync;
>>  } MultiFDRecvParams;
>>  
>> +typedef struct {
>> +    /* Prepare the send packet */
>> +    int (*send_prepare)(MultiFDSendParams *p, uint32_t used);
>> +    /* Write the send packet */
>> +    int (*send_write)(MultiFDSendParams *p, uint32_t used, Error **perr);
>> +    /* Read all pages */
>> +    int (*recv_pages)(MultiFDRecvParams *p, uint32_t used, Error **perr);
>> +} MultifdMethods;
>> +
>> +/* Multifd without compression */
>> +
>> +static int none_send_prepare(MultiFDSendParams *p, uint32_t used)
>> +{
>> +    p->next_packet_size = used * qemu_target_page_size();
>> +    return 0;
>> +}
>> +
>> +static int none_send_write(MultiFDSendParams *p, uint32_t used, Error **perr)
>> +{
>> +    return qio_channel_writev_all(p->c, p->pages->iov, used, perr);
>> +}
>> +
>> +static int none_recv_pages(MultiFDRecvParams *p, uint32_t used, Error **perr)
>> +{
>> +    return qio_channel_readv_all(p->c, p->pages->iov, used, perr);
>> +
>> +}
>> +
>> +MultifdMethods multifd_none_ops = {
>> +    .send_prepare = none_send_prepare,
>> +    .send_write = none_send_write,
>> +    .recv_pages = none_recv_pages
>> +};
>> +
>>  static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
>>  {
>>      MultiFDInit_t msg;
>> @@ -904,6 +938,8 @@ struct {
>>      uint64_t packet_num;
>>      /* send channels ready */
>>      QemuSemaphore channels_ready;
>> +    /* multifd ops */
>> +    MultifdMethods *ops;
>>  } *multifd_send_state;
>>  
>>  /*
>> @@ -1093,6 +1129,8 @@ static void *multifd_send_thread(void *opaque)
>>      /* initial packet */
>>      p->num_packets = 1;
>>  
>> +    multifd_send_state->ops = &multifd_none_ops;
>> +
>
> I agree with Wei Yang that is a bad idea; that should be done once
> before the first thread is started.
>
> Dave
>
>>      while (true) {
>>          qemu_sem_wait(&p->sem);
>>          qemu_mutex_lock(&p->mutex);
>> @@ -1102,7 +1140,12 @@ static void *multifd_send_thread(void *opaque)
>>              uint64_t packet_num = p->packet_num;
>>              uint32_t flags = p->flags;
>>  
>> -            p->next_packet_size = used * qemu_target_page_size();
>> +            if (used) {
>> +                ret = multifd_send_state->ops->send_prepare(p, used);
>> +                if (ret != 0) {
>> +                    break;
>> +                }
>> +            }
>>              multifd_send_fill_packet(p);
>>              p->flags = 0;
>>              p->num_packets++;
>> @@ -1120,8 +1163,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;
>>                  }
>> @@ -1223,6 +1265,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)
>> @@ -1324,6 +1368,7 @@ static void *multifd_recv_thread(void *opaque)
>>      trace_multifd_recv_thread_start(p->id);
>>      rcu_register_thread();
>>  
>> +    multifd_recv_state->ops = &multifd_none_ops;
>>      while (true) {
>>          uint32_t used;
>>          uint32_t flags;
>> @@ -1353,8 +1398,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;
>>              }
>> -- 
>> 2.21.0
>> 
> --
> Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK