[Qemu-devel] [PATCH v5 0/12] migration: Add a new feature to do live migration

[Qemu-devel] [PATCH v5 0/12] migration: Add a new feature to do live migration

[Liang Li]

This feature can help to reduce the data transferred about 60%, and the
migration time can also be reduced about 70%.

    Summary of changed from v4->v5
    
    -Fix some typo errors 
    -Fix the bug related with XBZRLE
    -Added some comments
    -Squash setting and querying commands into one patch
    -Fix the issue when doing migrate_cancel

 arch_init.c                       | 485 ++++++++++++++++++++++++++++++++++++--
 docs/multi-thread-compression.txt | 149 ++++++++++++
 hmp-commands.hx                   |  17 ++
 hmp.c                             |  56 +++++
 hmp.h                             |   4 +
 include/migration/migration.h     |  11 +
 include/migration/qemu-file.h     |   3 +
 migration/migration.c             | 130 ++++++++++
 migration/qemu-file.c             |  39 +++
 monitor.c                         |  25 ++
 qapi-schema.json                  |  93 +++++++-
 qmp-commands.hx                   |  66 ++++++
 12 files changed, 1056 insertions(+), 22 deletions(-)
 create mode 100644 docs/multi-thread-compression.txt

[Qemu-devel] [v5 01/12] docs: Add a doc about multiple thread compression

[Liang Li]

Give some details about the multiple thread (de)compression and
how to use it in live migration.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 docs/multi-thread-compression.txt | 149 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 149 insertions(+)
 create mode 100644 docs/multi-thread-compression.txt

diff --git a/docs/multi-thread-compression.txt b/docs/multi-thread-compression.txt
new file mode 100644
index 0000000..0d4d212
--- /dev/null
+++ b/docs/multi-thread-compression.txt
@@ -0,0 +1,149 @@
+Use multiple thread (de)compression in live migration
+=====================================================
+Copyright (C) 2015 Intel Corporation
+Author: Liang Li <liang.z.li@intel.com>
+
+This work is licensed under the terms of the GNU GPLv2 or later. See
+the COPYING file in the top-level directory.
+
+Contents:
+=========
+* Introduction
+* When to use
+* Performance
+* Usage
+* TODO
+
+Introduction
+============
+Instead of sending the guest memory directly, this solution will
+compress the RAM page before sending; after receiving, the data will
+be decompressed. Using compression in live migration can help
+to reduce the data transferred about 60%, this is very useful when the
+bandwidth is limited, and the total migration time can also be reduced
+about 70% in a typical case. In addition to this, the VM downtime can be
+reduced about 50%. The benefit depends on data's compressibility in VM.
+
+The process of compression will consume additional CPU cycles, and the
+extra CPU cycles will increase the migration time. On the other hand,
+the amount of data transferred will decrease; this factor can reduce
+the total migration time. If the process of the compression is quick
+enough, then the total migration time can be reduced, and multiple
+thread compression can be used to accelerate the compression process.
+
+The decompression speed of Zlib is at least 4 times as quick as
+compression, if the source and destination CPU have equal speed,
+keeping the compression thread count 4 times the decompression
+thread count can avoid CPU waste.
+
+Compression level can be used to control the compression speed and the
+compression ratio. High compression ratio will take more time, level 0
+stands for no compression, level 1 stands for the best compression
+speed, and level 9 stands for the best compression ratio. Users can
+select a level number between 0 and 9.
+
+
+When to use the multiple thread compression in live migration
+=============================================================
+Compression of data will consume extra CPU cycles; so in a system with
+high overhead of CPU, avoid using this feature. When the network
+bandwidth is very limited and the CPU resource is adequate, use of
+multiple thread compression will be very helpful. If both the CPU and
+the network bandwidth are adequate, use of multiple thread compression
+can still help to reduce the migration time.
+
+Performance
+===========
+Test environment:
+
+CPU: Intel(R) Xeon(R) CPU E5-2680 0 @ 2.70GHz
+Socket Count: 2
+RAM: 128G
+NIC: Intel I350 (10/100/1000Mbps)
+Host OS: CentOS 7 64-bit
+Guest OS: RHEL 6.5 64-bit
+Parameter: qemu-system-x86_64 -enable-kvm -smp 4 -m 4096
+ /share/ia32e_rhel6u5.qcow -monitor stdio
+
+There is no additional application is running on the guest when doing
+the test.
+
+
+Speed limit: 1000Gb/s
+---------------------------------------------------------------
+                    | original  | compress thread: 8
+                    |   way     | decompress thread: 2
+                    |           | compression level: 1
+---------------------------------------------------------------
+total time(msec):   |   3333    |  1833
+---------------------------------------------------------------
+downtime(msec):     |    100    |   27
+---------------------------------------------------------------
+transferred ram(kB):|  363536   | 107819
+---------------------------------------------------------------
+throughput(mbps):   |  893.73   | 482.22
+---------------------------------------------------------------
+total ram(kB):      |  4211524  | 4211524
+---------------------------------------------------------------
+
+There is an application running on the guest which write random numbers
+to RAM block areas periodically.
+
+Speed limit: 1000Gb/s
+---------------------------------------------------------------
+                    | original  | compress thread: 8
+                    |   way     | decompress thread: 2
+                    |           | compression level: 1
+---------------------------------------------------------------
+total time(msec):   |   37369   | 15989
+---------------------------------------------------------------
+downtime(msec):     |    337    |  173
+---------------------------------------------------------------
+transferred ram(kB):|  4274143  | 1699824
+---------------------------------------------------------------
+throughput(mbps):   |  936.99   | 870.95
+---------------------------------------------------------------
+total ram(kB):      |  4211524  | 4211524
+---------------------------------------------------------------
+
+Usage
+=====
+1. Verify both the source and destination QEMU are able
+to support the multiple thread compression migration:
+    {qemu} info_migrate_capabilities
+    {qemu} ... compress: off ...
+
+2. Activate compression on the source:
+    {qemu} migrate_set_capability compress on
+
+3. Set the compression thread count on source:
+    {qemu} migrate_set_parameter compress_threads 12
+
+4. Set the compression level on the source:
+    {qemu} migrate_set_parameter compress_level 1
+
+5. Set the decompression thread count on destination:
+    {qemu} migrate_set_parameter decompress_threads 3
+
+6. Start outgoing migration:
+    {qemu} migrate -d tcp:destination.host:4444
+    {qemu} info migrate
+    Capabilities: ... compress: on
+    ...
+
+The following are the default settings:
+    compress: off
+    compress_threads: 8
+    decompress_threads: 2
+    compress_level: 1 (which means best speed)
+
+So, only the first two steps are required to use the multiple
+thread compression in migration. You can do more if the default
+settings are not appropriate.
+
+TODO
+====
+Some faster (de)compression method such as LZ4 and Quicklz can help
+to reduce the CPU consumption when doing (de)compression. If using
+these faster (de)compression method, less (de)compression threads
+are needed when doing the migration.
-- 
1.9.1

[Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Liang Li]

Add the code to create and destroy the multiple threads those will
be used to do data compression. Left some functions empty to keep
clearness, and the code will be added later.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 arch_init.c                   | 79 ++++++++++++++++++++++++++++++++++++++++++-
 include/migration/migration.h |  9 +++++
 migration/migration.c         | 37 ++++++++++++++++++++
 3 files changed, 124 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 89c8fa4..709036c 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -332,6 +332,68 @@ static uint64_t migration_dirty_pages;
 static uint32_t last_version;
 static bool ram_bulk_stage;
 
+struct CompressParam {
+    /* To be done */
+};
+typedef struct CompressParam CompressParam;
+
+static CompressParam *comp_param;
+static bool quit_comp_thread;
+
+static void *do_data_compress(void *opaque)
+{
+    while (!quit_comp_thread) {
+
+    /* To be done */
+
+    }
+
+    return NULL;
+}
+
+static inline void terminate_compression_threads(void)
+{
+    quit_comp_thread = true;
+
+    /* To be done */
+}
+
+void migrate_compress_threads_join(MigrationState *s)
+{
+    int i, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    terminate_compression_threads();
+    thread_count = migrate_compress_threads();
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_join(s->compress_thread + i);
+    }
+    g_free(s->compress_thread);
+    g_free(comp_param);
+    s->compress_thread = NULL;
+    comp_param = NULL;
+}
+
+void migrate_compress_threads_create(MigrationState *s)
+{
+    int i, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    quit_comp_thread = false;
+    thread_count = migrate_compress_threads();
+    s->compress_thread = g_new0(QemuThread, thread_count);
+    comp_param = g_new0(CompressParam, thread_count);
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_create(s->compress_thread + i, "compress",
+                           do_data_compress, comp_param + i,
+                           QEMU_THREAD_JOINABLE);
+    }
+}
+
 /* Update the xbzrle cache to reflect a page that's been sent as all 0.
  * The important thing is that a stale (not-yet-0'd) page be replaced
  * by the new data.
@@ -645,6 +707,16 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
     return bytes_sent;
 }
 
+static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
+                                    ram_addr_t offset, bool last_stage)
+{
+    int bytes_sent = -1;
+
+    /* To be done*/
+
+    return bytes_sent;
+}
+
 /*
  * ram_find_and_save_block: Finds a page to send and sends it to f
  *
@@ -679,7 +751,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
                 ram_bulk_stage = false;
             }
         } else {
-            bytes_sent = ram_save_page(f, block, offset, last_stage);
+            if (migrate_use_compression()) {
+                bytes_sent = ram_save_compressed_page(f, block, offset,
+                                                      last_stage);
+            } else {
+                bytes_sent = ram_save_page(f, block, offset, last_stage);
+            }
 
             /* if page is unmodified, continue to the next */
             if (bytes_sent > 0) {
diff --git a/include/migration/migration.h b/include/migration/migration.h
index f37348b..228badb 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -50,6 +50,9 @@ struct MigrationState
     QemuThread thread;
     QEMUBH *cleanup_bh;
     QEMUFile *file;
+    QemuThread *compress_thread;
+    int compress_thread_count;
+    int compress_level;
 
     int state;
     MigrationParams params;
@@ -108,6 +111,8 @@ bool migration_has_finished(MigrationState *);
 bool migration_has_failed(MigrationState *);
 MigrationState *migrate_get_current(void);
 
+void migrate_compress_threads_create(MigrationState *s);
+void migrate_compress_threads_join(MigrationState *s);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -157,6 +162,10 @@ int64_t migrate_xbzrle_cache_size(void);
 
 int64_t xbzrle_cache_resize(int64_t new_size);
 
+bool migrate_use_compression(void);
+int migrate_compress_level(void);
+int migrate_compress_threads(void);
+
 void ram_control_before_iterate(QEMUFile *f, uint64_t flags);
 void ram_control_after_iterate(QEMUFile *f, uint64_t flags);
 void ram_control_load_hook(QEMUFile *f, uint64_t flags);
diff --git a/migration/migration.c b/migration/migration.c
index b3adbc6..309443e 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -43,6 +43,11 @@ enum {
 #define BUFFER_DELAY     100
 #define XFER_LIMIT_RATIO (1000 / BUFFER_DELAY)
 
+/* Default compression thread count */
+#define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+/*0: means nocompress, 1: best speed, ... 9: best compress ratio */
+#define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
+
 /* Migration XBZRLE default cache size */
 #define DEFAULT_MIGRATE_CACHE_SIZE (64 * 1024 * 1024)
 
@@ -60,6 +65,8 @@ MigrationState *migrate_get_current(void)
         .bandwidth_limit = MAX_THROTTLE,
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
+        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
     };
 
     return &current_migration;
@@ -302,6 +309,7 @@ static void migrate_fd_cleanup(void *opaque)
         qemu_thread_join(&s->thread);
         qemu_mutex_lock_iothread();
 
+        migrate_compress_threads_join(s);
         qemu_fclose(s->file);
         s->file = NULL;
     }
@@ -385,6 +393,8 @@ static MigrationState *migrate_init(const MigrationParams *params)
     int64_t bandwidth_limit = s->bandwidth_limit;
     bool enabled_capabilities[MIGRATION_CAPABILITY_MAX];
     int64_t xbzrle_cache_size = s->xbzrle_cache_size;
+    int compress_level = s->compress_level;
+    int compress_thread_count = s->compress_thread_count;
 
     memcpy(enabled_capabilities, s->enabled_capabilities,
            sizeof(enabled_capabilities));
@@ -395,6 +405,8 @@ static MigrationState *migrate_init(const MigrationParams *params)
            sizeof(enabled_capabilities));
     s->xbzrle_cache_size = xbzrle_cache_size;
 
+    s->compress_level = compress_level;
+    s->compress_thread_count = compress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -567,6 +579,30 @@ bool migrate_zero_blocks(void)
     return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS];
 }
 
+bool migrate_use_compression(void)
+{
+    /* Disable compression before the patch series are applied */
+    return false;
+}
+
+int migrate_compress_level(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->compress_level;
+}
+
+int migrate_compress_threads(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->compress_thread_count;
+}
+
 int migrate_use_xbzrle(void)
 {
     MigrationState *s;
@@ -707,6 +743,7 @@ void migrate_fd_connect(MigrationState *s)
     /* Notify before starting migration thread */
     notifier_list_notify(&migration_state_notifiers, s);
 
+    migrate_compress_threads_create(s);
     qemu_thread_create(&s->thread, "migration", migration_thread, s,
                        QEMU_THREAD_JOINABLE);
 }
-- 
1.9.1

[Qemu-devel] [v5 03/12] migration: Add the framework of multi-thread decompression

[Liang Li]

Add the code to create and destroy the multiple threads those will be
used to do data decompression. Left some functions empty just to keep
clearness, and the code will be added later.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 arch_init.c                   | 76 +++++++++++++++++++++++++++++++++++++++++++
 include/migration/migration.h |  4 +++
 migration/migration.c         | 18 ++++++++++
 3 files changed, 98 insertions(+)

diff --git a/arch_init.c b/arch_init.c
index 709036c..bbb584c 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -24,6 +24,7 @@
 #include <stdint.h>
 #include <stdarg.h>
 #include <stdlib.h>
+#include <zlib.h>
 #ifndef _WIN32
 #include <sys/types.h>
 #include <sys/mman.h>
@@ -126,6 +127,7 @@ static uint64_t bitmap_sync_count;
 #define RAM_SAVE_FLAG_CONTINUE 0x20
 #define RAM_SAVE_FLAG_XBZRLE   0x40
 /* 0x80 is reserved in migration.h start with 0x100 next */
+#define RAM_SAVE_FLAG_COMPRESS_PAGE    0x100
 
 static struct defconfig_file {
     const char *filename;
@@ -337,8 +339,17 @@ struct CompressParam {
 };
 typedef struct CompressParam CompressParam;
 
+struct DecompressParam {
+    /* To be done */
+};
+typedef struct DecompressParam DecompressParam;
+
 static CompressParam *comp_param;
 static bool quit_comp_thread;
+static bool quit_decomp_thread;
+static DecompressParam *decomp_param;
+static QemuThread *decompress_threads;
+static uint8_t *compressed_data_buf;
 
 static void *do_data_compress(void *opaque)
 {
@@ -1128,10 +1139,58 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
     }
 }
 
+static void *do_data_decompress(void *opaque)
+{
+    while (!quit_decomp_thread) {
+        /* To be done */
+    }
+
+    return NULL;
+}
+
+void migrate_decompress_threads_create(int count)
+{
+    int i;
+
+    decompress_threads = g_new0(QemuThread, count);
+    decomp_param = g_new0(DecompressParam, count);
+    compressed_data_buf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
+    quit_decomp_thread = false;
+    for (i = 0; i < count; i++) {
+        qemu_thread_create(decompress_threads + i, "decompress",
+                           do_data_decompress, decomp_param + i,
+                           QEMU_THREAD_JOINABLE);
+    }
+}
+
+void migrate_decompress_threads_join(void)
+{
+    int i, thread_count;
+
+    quit_decomp_thread = true;
+    thread_count = migrate_decompress_threads();
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_join(decompress_threads + i);
+    }
+    g_free(decompress_threads);
+    g_free(decomp_param);
+    g_free(compressed_data_buf);
+    decompress_threads = NULL;
+    decomp_param = NULL;
+    compressed_data_buf = NULL;
+}
+
+static void decompress_data_with_multi_threads(uint8_t *compbuf,
+                                               void *host, int len)
+{
+    /* To be done */
+}
+
 static int ram_load(QEMUFile *f, void *opaque, int version_id)
 {
     int flags = 0, ret = 0;
     static uint64_t seq_iter;
+    int len = 0;
 
     seq_iter++;
 
@@ -1208,6 +1267,23 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
 
             qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
             break;
+        case RAM_SAVE_FLAG_COMPRESS_PAGE:
+            host = host_from_stream_offset(f, addr, flags);
+            if (!host) {
+                error_report("Invalid RAM offset " RAM_ADDR_FMT, addr);
+                ret = -EINVAL;
+                break;
+            }
+
+            len = qemu_get_be32(f);
+            if (len < 0 || len > compressBound(TARGET_PAGE_SIZE)) {
+                error_report("Invalid compressed data length: %d", len);
+                ret = -EINVAL;
+                break;
+            }
+            qemu_get_buffer(f, compressed_data_buf, len);
+            decompress_data_with_multi_threads(compressed_data_buf, host, len);
+            break;
         case RAM_SAVE_FLAG_XBZRLE:
             host = host_from_stream_offset(f, addr, flags);
             if (!host) {
diff --git a/include/migration/migration.h b/include/migration/migration.h
index 228badb..9ac1b23 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -52,6 +52,7 @@ struct MigrationState
     QEMUFile *file;
     QemuThread *compress_thread;
     int compress_thread_count;
+    int decompress_thread_count;
     int compress_level;
 
     int state;
@@ -113,6 +114,8 @@ MigrationState *migrate_get_current(void);
 
 void migrate_compress_threads_create(MigrationState *s);
 void migrate_compress_threads_join(MigrationState *s);
+void migrate_decompress_threads_create(int count);
+void migrate_decompress_threads_join(void);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -165,6 +168,7 @@ int64_t xbzrle_cache_resize(int64_t new_size);
 bool migrate_use_compression(void);
 int migrate_compress_level(void);
 int migrate_compress_threads(void);
+int migrate_decompress_threads(void);
 
 void ram_control_before_iterate(QEMUFile *f, uint64_t flags);
 void ram_control_after_iterate(QEMUFile *f, uint64_t flags);
diff --git a/migration/migration.c b/migration/migration.c
index 309443e..8e15a79 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -45,6 +45,7 @@ enum {
 
 /* Default compression thread count */
 #define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+#define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
 /*0: means nocompress, 1: best speed, ... 9: best compress ratio */
 #define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
 
@@ -66,6 +67,7 @@ MigrationState *migrate_get_current(void)
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
         .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
         .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
     };
 
@@ -106,6 +108,7 @@ static void process_incoming_migration_co(void *opaque)
     free_xbzrle_decoded_buf();
     if (ret < 0) {
         error_report("load of migration failed: %s", strerror(-ret));
+        migrate_decompress_threads_join();
         exit(EXIT_FAILURE);
     }
     qemu_announce_self();
@@ -115,6 +118,7 @@ static void process_incoming_migration_co(void *opaque)
     if (local_err) {
         qerror_report_err(local_err);
         error_free(local_err);
+        migrate_decompress_threads_join();
         exit(EXIT_FAILURE);
     }
 
@@ -123,12 +127,15 @@ static void process_incoming_migration_co(void *opaque)
     } else {
         runstate_set(RUN_STATE_PAUSED);
     }
+    migrate_decompress_threads_join();
 }
 
 void process_incoming_migration(QEMUFile *f)
 {
     Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
     int fd = qemu_get_fd(f);
+    int thread_count = migrate_decompress_threads();
+    migrate_decompress_threads_create(thread_count);
 
     assert(fd != -1);
     qemu_set_nonblock(fd);
@@ -395,6 +402,7 @@ static MigrationState *migrate_init(const MigrationParams *params)
     int64_t xbzrle_cache_size = s->xbzrle_cache_size;
     int compress_level = s->compress_level;
     int compress_thread_count = s->compress_thread_count;
+    int decompress_thread_count = s->decompress_thread_count;
 
     memcpy(enabled_capabilities, s->enabled_capabilities,
            sizeof(enabled_capabilities));
@@ -407,6 +415,7 @@ static MigrationState *migrate_init(const MigrationParams *params)
 
     s->compress_level = compress_level;
     s->compress_thread_count = compress_thread_count;
+    s->decompress_thread_count = decompress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -603,6 +612,15 @@ int migrate_compress_threads(void)
     return s->compress_thread_count;
 }
 
+int migrate_decompress_threads(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->decompress_thread_count;
+}
+
 int migrate_use_xbzrle(void)
 {
     MigrationState *s;
-- 
1.9.1

[Qemu-devel] [v5 04/12] qemu-file: Add compression functions to QEMUFile

[Liang Li]

qemu_put_compression_data() compress the data and put it to QEMUFile.
qemu_put_qemu_file() put the data in the buffer of source QEMUFile to
destination QEMUFile.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 include/migration/qemu-file.h |  3 +++
 migration/qemu-file.c         | 39 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/include/migration/qemu-file.h b/include/migration/qemu-file.h
index a923cec..52192ad 100644
--- a/include/migration/qemu-file.h
+++ b/include/migration/qemu-file.h
@@ -160,6 +160,9 @@ void qemu_put_be32(QEMUFile *f, unsigned int v);
 void qemu_put_be64(QEMUFile *f, uint64_t v);
 int qemu_peek_buffer(QEMUFile *f, uint8_t *buf, int size, size_t offset);
 int qemu_get_buffer(QEMUFile *f, uint8_t *buf, int size);
+ssize_t qemu_put_compression_data(QEMUFile *f, const uint8_t *p, size_t size,
+                                  int level);
+int qemu_put_qemu_file(QEMUFile *f_des, QEMUFile *f_src);
 /*
  * Note that you can only peek continuous bytes from where the current pointer
  * is; you aren't guaranteed to be able to peak to +n bytes unless you've
diff --git a/migration/qemu-file.c b/migration/qemu-file.c
index e66e557..3e54f6d 100644
--- a/migration/qemu-file.c
+++ b/migration/qemu-file.c
@@ -21,6 +21,7 @@
  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  * THE SOFTWARE.
  */
+#include <zlib.h>
 #include "qemu-common.h"
 #include "qemu/iov.h"
 #include "qemu/sockets.h"
@@ -545,3 +546,41 @@ uint64_t qemu_get_be64(QEMUFile *f)
     v |= qemu_get_be32(f);
     return v;
 }
+
+/* compress size bytes of data start at p with specific compression
+ * level and store the compressed data to the buffer of f.
+ */
+
+ssize_t qemu_put_compression_data(QEMUFile *f, const uint8_t *p, size_t size,
+                                  int level)
+{
+    ssize_t blen = IO_BUF_SIZE - f->buf_index - sizeof(int32_t);
+
+    if (blen < compressBound(size)) {
+        return 0;
+    }
+    if (compress2(f->buf + f->buf_index + sizeof(int32_t), (uLongf *)&blen,
+                  (Bytef *)p, size, level) != Z_OK) {
+        error_report("Compress Failed!");
+        return 0;
+    }
+    qemu_put_be32(f, blen);
+    f->buf_index += blen;
+    return blen + sizeof(int32_t);
+}
+
+/* Put the data in the buffer of f_src to the buffer of f_des, and
+ * then reset the buf_index of f_src to 0.
+ */
+
+int qemu_put_qemu_file(QEMUFile *f_des, QEMUFile *f_src)
+{
+    int len = 0;
+
+    if (f_src->buf_index > 0) {
+        len = f_src->buf_index;
+        qemu_put_buffer(f_des, f_src->buf, f_src->buf_index);
+        f_src->buf_index = 0;
+    }
+    return len;
+}
-- 
1.9.1

[Qemu-devel] [v5 05/12] arch_init: Alloc and free data struct for compression

[Liang Li]

Define the data structure and variables used to do multiple thread
compression, and add the code to initialize and free them.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 arch_init.c | 34 +++++++++++++++++++++++++++++++++-
 1 file changed, 33 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index bbb584c..7ccad8b 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -335,7 +335,12 @@ static uint32_t last_version;
 static bool ram_bulk_stage;
 
 struct CompressParam {
-    /* To be done */
+    bool busy;
+    QEMUFile *file;
+    QemuMutex mutex;
+    QemuCond cond;
+    RAMBlock *block;
+    ram_addr_t offset;
 };
 typedef struct CompressParam CompressParam;
 
@@ -345,6 +350,14 @@ struct DecompressParam {
 typedef struct DecompressParam DecompressParam;
 
 static CompressParam *comp_param;
+/* comp_done_cond is used to wake up the migration thread when
+ * one of the compression threads has finished the compression.
+ * comp_done_lock is used to co-work with comp_done_cond.
+ */
+static QemuMutex *comp_done_lock;
+static QemuCond *comp_done_cond;
+/* The empty QEMUFileOps will be used by file in CompressParam */
+static const QEMUFileOps empty_ops = { };
 static bool quit_comp_thread;
 static bool quit_decomp_thread;
 static DecompressParam *decomp_param;
@@ -380,11 +393,20 @@ void migrate_compress_threads_join(MigrationState *s)
     thread_count = migrate_compress_threads();
     for (i = 0; i < thread_count; i++) {
         qemu_thread_join(s->compress_thread + i);
+        qemu_fclose(comp_param[i].file);
+        qemu_mutex_destroy(&comp_param[i].mutex);
+        qemu_cond_destroy(&comp_param[i].cond);
     }
+    qemu_mutex_destroy(comp_done_lock);
+    qemu_cond_destroy(comp_done_cond);
     g_free(s->compress_thread);
     g_free(comp_param);
+    g_free(comp_done_cond);
+    g_free(comp_done_lock);
     s->compress_thread = NULL;
     comp_param = NULL;
+    comp_done_cond = NULL;
+    comp_done_lock = NULL;
 }
 
 void migrate_compress_threads_create(MigrationState *s)
@@ -398,7 +420,17 @@ void migrate_compress_threads_create(MigrationState *s)
     thread_count = migrate_compress_threads();
     s->compress_thread = g_new0(QemuThread, thread_count);
     comp_param = g_new0(CompressParam, thread_count);
+    comp_done_cond = g_new0(QemuCond, 1);
+    comp_done_lock = g_new0(QemuMutex, 1);
+    qemu_cond_init(comp_done_cond);
+    qemu_mutex_init(comp_done_lock);
     for (i = 0; i < thread_count; i++) {
+        /* com_param[i].file is just used as a dummy buffer to save data, set
+         * it's ops to empty.
+         */
+        comp_param[i].file = qemu_fopen_ops(NULL, &empty_ops);
+        qemu_mutex_init(&comp_param[i].mutex);
+        qemu_cond_init(&comp_param[i].cond);
         qemu_thread_create(s->compress_thread + i, "compress",
                            do_data_compress, comp_param + i,
                            QEMU_THREAD_JOINABLE);
-- 
1.9.1

[Qemu-devel] [v5 06/12] arch_init: Add and free data struct for decompression

[Liang Li]

Define the data structure and variables used to do multiple thread
decompression, and add the code to initialize and free them.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 arch_init.c | 13 ++++++++++++-
 1 file changed, 12 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 7ccad8b..66f4bc1 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -345,7 +345,12 @@ struct CompressParam {
 typedef struct CompressParam CompressParam;
 
 struct DecompressParam {
-    /* To be done */
+    bool busy;
+    QemuMutex mutex;
+    QemuCond cond;
+    void *des;
+    uint8 *compbuf;
+    int len;
 };
 typedef struct DecompressParam DecompressParam;
 
@@ -1189,6 +1194,9 @@ void migrate_decompress_threads_create(int count)
     compressed_data_buf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
     quit_decomp_thread = false;
     for (i = 0; i < count; i++) {
+        qemu_mutex_init(&decomp_param[i].mutex);
+        qemu_cond_init(&decomp_param[i].cond);
+        decomp_param[i].compbuf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
         qemu_thread_create(decompress_threads + i, "decompress",
                            do_data_decompress, decomp_param + i,
                            QEMU_THREAD_JOINABLE);
@@ -1203,6 +1211,9 @@ void migrate_decompress_threads_join(void)
     thread_count = migrate_decompress_threads();
     for (i = 0; i < thread_count; i++) {
         qemu_thread_join(decompress_threads + i);
+        qemu_mutex_destroy(&decomp_param[i].mutex);
+        qemu_cond_destroy(&decomp_param[i].cond);
+        g_free(decomp_param[i].compbuf);
     }
     g_free(decompress_threads);
     g_free(decomp_param);
-- 
1.9.1

[Qemu-devel] [v5 07/12] migration: Split the function ram_save_page

[Liang Li]

Split the function ram_save_page for code reuse purpose.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 arch_init.c | 51 +++++++++++++++++++++++++++++++++------------------
 1 file changed, 33 insertions(+), 18 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 66f4bc1..fe062db 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -681,12 +681,28 @@ static void migration_bitmap_sync(void)
     }
 }
 
+static int save_zero_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset,
+                          uint8_t *p, int cont)
+{
+    int bytes_sent = -1;
+
+    if (is_zero_range(p, TARGET_PAGE_SIZE)) {
+        acct_info.dup_pages++;
+        bytes_sent = save_block_hdr(f, block, offset, cont,
+                                    RAM_SAVE_FLAG_COMPRESS);
+        qemu_put_byte(f, 0);
+        bytes_sent++;
+    }
+
+    return bytes_sent;
+}
+
 /*
  * ram_save_page: Send the given page to the stream
  *
  * Returns: Number of bytes written.
  */
-static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
+static int ram_save_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset,
                          bool last_stage)
 {
     int bytes_sent;
@@ -717,24 +733,23 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
                 acct_info.dup_pages++;
             }
         }
-    } else if (is_zero_range(p, TARGET_PAGE_SIZE)) {
-        acct_info.dup_pages++;
-        bytes_sent = save_block_hdr(f, block, offset, cont,
-                                    RAM_SAVE_FLAG_COMPRESS);
-        qemu_put_byte(f, 0);
-        bytes_sent++;
-        /* Must let xbzrle know, otherwise a previous (now 0'd) cached
-         * page would be stale
-         */
-        xbzrle_cache_zero_page(current_addr);
-    } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
-        bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
-                                      offset, cont, last_stage);
-        if (!last_stage) {
-            /* Can't send this cached data async, since the cache page
-             * might get updated before it gets to the wire
+    } else {
+        bytes_sent = save_zero_page(f, block, offset, p, cont);
+        if (bytes_sent > 0) {
+
+            /* Must let xbzrle know, otherwise a previous (now 0'd) cached
+             * page would be stale
              */
-            send_async = false;
+            xbzrle_cache_zero_page(current_addr);
+        } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
+            bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
+                                          offset, cont, last_stage);
+            if (!last_stage) {
+                /* Can't send this cached data async, since the cache page
+                 * might get updated before it gets to the wire
+                 */
+                send_async = false;
+            }
         }
     }
 
-- 
1.9.1

[Qemu-devel] [v5 08/12] migration: Add the core code of multi-thread compression

[Liang Li]

Implement the core logic of the multiple thread compression. At this
point, multiple thread compression can't co-work with xbzrle yet.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 arch_init.c | 193 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 185 insertions(+), 8 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index fe062db..17b7f15 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -363,18 +363,44 @@ static QemuMutex *comp_done_lock;
 static QemuCond *comp_done_cond;
 /* The empty QEMUFileOps will be used by file in CompressParam */
 static const QEMUFileOps empty_ops = { };
+
+/* one_byte_count is used to count the bytes that is added to
+ * bytes_transferred but not actually transferred, at the proper
+ * time, we should sub one_byte_count from bytes_transferred to
+ * make bytes_transferred accurate.
+ */
+static int one_byte_count;
 static bool quit_comp_thread;
 static bool quit_decomp_thread;
 static DecompressParam *decomp_param;
 static QemuThread *decompress_threads;
 static uint8_t *compressed_data_buf;
 
+static int do_compress_ram_page(CompressParam *param);
+
 static void *do_data_compress(void *opaque)
 {
-    while (!quit_comp_thread) {
-
-    /* To be done */
+    CompressParam *param = opaque;
 
+    while (!quit_comp_thread) {
+        qemu_mutex_lock(&param->mutex);
+        /* Re-check the quit_comp_thread in case of
+         * terminate_compression_threads is called just before
+         * qemu_mutex_lock(&param->mutex) and after
+         * while(!quit_comp_thread), re-check it here can make
+         * sure the compression thread terminate as expected.
+         */
+        while (!param->busy && !quit_comp_thread) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+        }
+        qemu_mutex_unlock(&param->mutex);
+        if (!quit_comp_thread) {
+            do_compress_ram_page(param);
+        }
+        qemu_mutex_lock(comp_done_lock);
+        param->busy = false;
+        qemu_cond_signal(comp_done_cond);
+        qemu_mutex_unlock(comp_done_lock);
     }
 
     return NULL;
@@ -382,9 +408,15 @@ static void *do_data_compress(void *opaque)
 
 static inline void terminate_compression_threads(void)
 {
-    quit_comp_thread = true;
+    int idx, thread_count;
 
-    /* To be done */
+    thread_count = migrate_compress_threads();
+    quit_comp_thread = true;
+    for (idx = 0; idx < thread_count; idx++) {
+        qemu_mutex_lock(&comp_param[idx].mutex);
+        qemu_cond_signal(&comp_param[idx].cond);
+        qemu_mutex_unlock(&comp_param[idx].mutex);
+    }
 }
 
 void migrate_compress_threads_join(MigrationState *s)
@@ -770,12 +802,157 @@ static int ram_save_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset,
     return bytes_sent;
 }
 
+static int do_compress_ram_page(CompressParam *param)
+{
+    int bytes_sent, cont;
+    int blen;
+    uint8_t *p;
+    RAMBlock *block = param->block;
+    ram_addr_t offset = param->offset;
+
+    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
+    p = memory_region_get_ram_ptr(block->mr) + offset;
+
+    bytes_sent = save_block_hdr(param->file, block, offset, cont,
+                                RAM_SAVE_FLAG_COMPRESS_PAGE);
+    blen = qemu_put_compression_data(param->file, p, TARGET_PAGE_SIZE,
+                                     migrate_compress_level());
+    bytes_sent += blen;
+    atomic_inc(&acct_info.norm_pages);
+
+    return bytes_sent;
+}
+
+static inline void start_compression(CompressParam *param)
+{
+    qemu_mutex_lock(&param->mutex);
+    param->busy = true;
+    qemu_cond_signal(&param->cond);
+    qemu_mutex_unlock(&param->mutex);
+}
+
+
+static uint64_t bytes_transferred;
+
+static void flush_compressed_data(QEMUFile *f)
+{
+    int idx, len, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    thread_count = migrate_compress_threads();
+    for (idx = 0; idx < thread_count; idx++) {
+        if (comp_param[idx].busy) {
+            qemu_mutex_lock(comp_done_lock);
+            while (comp_param[idx].busy && !quit_comp_thread) {
+                qemu_cond_wait(comp_done_cond, comp_done_lock);
+            }
+            qemu_mutex_unlock(comp_done_lock);
+        }
+        len = qemu_put_qemu_file(f, comp_param[idx].file);
+        bytes_transferred += len;
+    }
+    if ((one_byte_count > 0) && (bytes_transferred > one_byte_count)) {
+        bytes_transferred -= one_byte_count;
+        one_byte_count = 0;
+    }
+}
+
+static inline void set_compress_params(CompressParam *param, RAMBlock *block,
+                                       ram_addr_t offset)
+{
+    param->block = block;
+    param->offset = offset;
+}
+
+static int compress_page_with_multi_thread(QEMUFile *f, RAMBlock *block,
+                                           ram_addr_t offset)
+{
+    int idx, thread_count, bytes_sent = 0;
+
+    thread_count = migrate_compress_threads();
+    qemu_mutex_lock(comp_done_lock);
+    while (true) {
+        for (idx = 0; idx < thread_count; idx++) {
+            if (!comp_param[idx].busy) {
+                bytes_sent = qemu_put_qemu_file(f, comp_param[idx].file);
+                set_compress_params(&comp_param[idx], block, offset);
+                start_compression(&comp_param[idx]);
+                if (bytes_sent == 0) {
+                    /* set bytes_sent to 1 in this case to prevent migration
+                     * from terminating, this 1 byte will be added to
+                     * bytes_transferred later, minus 1 to keep the
+                     * bytes_transferred accurate */
+                    bytes_sent = 1;
+                    if (bytes_transferred <= 0) {
+                        one_byte_count++;
+                    } else {
+                        bytes_transferred -= 1;
+                    }
+                }
+                break;
+            }
+        }
+        if (bytes_sent > 0) {
+            break;
+        } else {
+            qemu_cond_wait(comp_done_cond, comp_done_lock);
+        }
+    }
+    qemu_mutex_unlock(comp_done_lock);
+
+    return bytes_sent;
+}
+
 static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
                                     ram_addr_t offset, bool last_stage)
 {
     int bytes_sent = -1;
+    MemoryRegion *mr = block->mr;
+    uint8_t *p;
+    int ret;
+    int cont;
 
-    /* To be done*/
+    p = memory_region_get_ram_ptr(mr) + offset;
+    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
+    ret = ram_control_save_page(f, block->offset,
+                                offset, TARGET_PAGE_SIZE, &bytes_sent);
+    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
+        if (ret != RAM_SAVE_CONTROL_DELAYED) {
+            if (bytes_sent > 0) {
+                acct_info.norm_pages++;
+            } else if (bytes_sent == 0) {
+                acct_info.dup_pages++;
+            }
+        }
+    } else {
+        /* When starting the process of a new block, the first page of
+         * the block should be sent out before other pages in the same
+         * block, and all the pages in last block should have been sent
+         * out, keeping this order is important, because the 'cont' flag
+         * is used to avoid resending the block name.
+         */
+        if (block != last_sent_block) {
+            flush_compressed_data(f);
+            bytes_sent = save_zero_page(f, block, offset, p, cont);
+            if (bytes_sent == -1) {
+                set_compress_params(&comp_param[0], block, offset);
+                /* Use the qemu thread to compress the data to make sure the
+                 * first page is sent out before other pages
+                 */
+                bytes_sent = do_compress_ram_page(&comp_param[0]);
+                if (bytes_sent > 0) {
+                    qemu_put_qemu_file(f, comp_param[0].file);
+                }
+            }
+        } else {
+            bytes_sent = save_zero_page(f, block, offset, p, cont);
+            if (bytes_sent == -1) {
+                bytes_sent = compress_page_with_multi_thread(f, block, offset);
+            }
+        }
+    }
 
     return bytes_sent;
 }
@@ -834,8 +1011,6 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
     return bytes_sent;
 }
 
-static uint64_t bytes_transferred;
-
 void acct_update_position(QEMUFile *f, size_t size, bool zero)
 {
     uint64_t pages = size / TARGET_PAGE_SIZE;
@@ -1043,6 +1218,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
         i++;
     }
 
+    flush_compressed_data(f);
     qemu_mutex_unlock_ramlist();
 
     /*
@@ -1089,6 +1265,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
         bytes_transferred += bytes_sent;
     }
 
+    flush_compressed_data(f);
     ram_control_after_iterate(f, RAM_CONTROL_FINISH);
     migration_end();
 
-- 
1.9.1

[Qemu-devel] [v5 09/12] migration: Make compression co-work with xbzrle

[Liang Li]

Now, multiple thread compression can co-work with xbzrle. when
xbzrle is on, multiple thread compression will only work at the
first round of RAM data sync.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
---
 arch_init.c | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 17b7f15..12dfa34 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -370,6 +370,7 @@ static const QEMUFileOps empty_ops = { };
  * make bytes_transferred accurate.
  */
 static int one_byte_count;
+static bool compression_switch;
 static bool quit_comp_thread;
 static bool quit_decomp_thread;
 static DecompressParam *decomp_param;
@@ -454,6 +455,7 @@ void migrate_compress_threads_create(MigrationState *s)
         return;
     }
     quit_comp_thread = false;
+    compression_switch = true;
     thread_count = migrate_compress_threads();
     s->compress_thread = g_new0(QemuThread, thread_count);
     comp_param = g_new0(CompressParam, thread_count);
@@ -989,9 +991,16 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
                 block = QTAILQ_FIRST(&ram_list.blocks);
                 complete_round = true;
                 ram_bulk_stage = false;
+                if (migrate_use_xbzrle()) {
+                    /* If xbzrle is on, stop using the data compression at this
+                     * point. In theory, xbzrle can do better than compression.
+                     */
+                    flush_compressed_data(f);
+                    compression_switch = false;
+                }
             }
         } else {
-            if (migrate_use_compression()) {
+            if (compression_switch && migrate_use_compression()) {
                 bytes_sent = ram_save_compressed_page(f, block, offset,
                                                       last_stage);
             } else {
-- 
1.9.1

[Qemu-devel] [v5 10/12] migration: Add the core code for decompression

[Liang Li]

Implement the core logic of multiple thread decompression,
the decompression can work now.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 arch_init.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 48 insertions(+), 2 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 12dfa34..f9e6a95 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -833,6 +833,13 @@ static inline void start_compression(CompressParam *param)
     qemu_mutex_unlock(&param->mutex);
 }
 
+static inline void start_decompression(DecompressParam *param)
+{
+    qemu_mutex_lock(&param->mutex);
+    param->busy = true;
+    qemu_cond_signal(&param->cond);
+    qemu_mutex_unlock(&param->mutex);
+}
 
 static uint64_t bytes_transferred;
 
@@ -1379,8 +1386,26 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
 
 static void *do_data_decompress(void *opaque)
 {
+    DecompressParam *param = opaque;
+    size_t pagesize;
+
     while (!quit_decomp_thread) {
-        /* To be done */
+        qemu_mutex_lock(&param->mutex);
+        while (!param->busy && !quit_decomp_thread) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+            pagesize = TARGET_PAGE_SIZE;
+            if (!quit_decomp_thread) {
+                /* uncompress() will return failed in some case, especially
+                 * when the page is dirted when doing the compression, it's
+                 * not a problem because the dirty page will be retransferred
+                 * and uncompress() won't break the data in other pages.
+                 */
+                uncompress((Bytef *)param->des, &pagesize,
+                           (const Bytef *)param->compbuf, param->len);
+            }
+            param->busy = false;
+        }
+        qemu_mutex_unlock(&param->mutex);
     }
 
     return NULL;
@@ -1411,6 +1436,11 @@ void migrate_decompress_threads_join(void)
     quit_decomp_thread = true;
     thread_count = migrate_decompress_threads();
     for (i = 0; i < thread_count; i++) {
+        qemu_mutex_lock(&decomp_param[i].mutex);
+        qemu_cond_signal(&decomp_param[i].cond);
+        qemu_mutex_unlock(&decomp_param[i].mutex);
+    }
+    for (i = 0; i < thread_count; i++) {
         qemu_thread_join(decompress_threads + i);
         qemu_mutex_destroy(&decomp_param[i].mutex);
         qemu_cond_destroy(&decomp_param[i].cond);
@@ -1427,7 +1457,23 @@ void migrate_decompress_threads_join(void)
 static void decompress_data_with_multi_threads(uint8_t *compbuf,
                                                void *host, int len)
 {
-    /* To be done */
+    int idx, thread_count;
+
+    thread_count = migrate_decompress_threads();
+    while (true) {
+        for (idx = 0; idx < thread_count; idx++) {
+            if (!decomp_param[idx].busy) {
+                memcpy(decomp_param[idx].compbuf, compbuf, len);
+                decomp_param[idx].des = host;
+                decomp_param[idx].len = len;
+                start_decompression(&decomp_param[idx]);
+                break;
+            }
+        }
+        if (idx < thread_count) {
+            break;
+        }
+    }
 }
 
 static int ram_load(QEMUFile *f, void *opaque, int version_id)
-- 
1.9.1

[Qemu-devel] [v5 11/12] migration: Add interface to control compression

[Liang Li]

The multiple compression threads can be turned on/off through
qmp and hmp interface before doing live migration.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Eric Blake <eblake@redhat.com>
---
 migration/migration.c | 7 +++++--
 qapi-schema.json      | 7 ++++++-
 2 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index 8e15a79..55f749e 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -590,8 +590,11 @@ bool migrate_zero_blocks(void)
 
 bool migrate_use_compression(void)
 {
-    /* Disable compression before the patch series are applied */
-    return false;
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->enabled_capabilities[MIGRATION_CAPABILITY_COMPRESS];
 }
 
 int migrate_compress_level(void)
diff --git a/qapi-schema.json b/qapi-schema.json
index e16f8eb..0dfc4ce 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -491,13 +491,18 @@
 #          to enable the capability on the source VM. The feature is disabled by
 #          default. (since 1.6)
 #
+# @compress: Use multiple compression threads to accelerate live migration.
+#          This feature can help to reduce the migration traffic, by sending
+#          compressed pages. The feature is disabled by default. (since 2.3)
+#
 # @auto-converge: If enabled, QEMU will automatically throttle down the guest
 #          to speed up convergence of RAM migration. (since 1.6)
 #
 # Since: 1.2
 ##
 { 'enum': 'MigrationCapability',
-  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks'] }
+  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks',
+           'compress'] }
 
 ##
 # @MigrationCapabilityStatus
-- 
1.9.1

[Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Liang Li]

Add the qmp and hmp commands to tune and query the parameters used in
live migration.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 hmp-commands.hx               | 17 ++++++++
 hmp.c                         | 56 +++++++++++++++++++++++++
 hmp.h                         |  4 ++
 include/migration/migration.h |  4 +-
 migration/migration.c         | 96 +++++++++++++++++++++++++++++++++++++------
 monitor.c                     | 25 +++++++++++
 qapi-schema.json              | 86 ++++++++++++++++++++++++++++++++++++++
 qmp-commands.hx               | 66 +++++++++++++++++++++++++++++
 8 files changed, 339 insertions(+), 15 deletions(-)

diff --git a/hmp-commands.hx b/hmp-commands.hx
index e37bc8b..ed0c06a 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -985,6 +985,21 @@ Enable/Disable the usage of a capability @var{capability} for migration.
 ETEXI
 
     {
+        .name       = "migrate_set_parameter",
+        .args_type  = "parameter:s,value:i",
+        .params     = "parameter value",
+        .help       = "Set the parameter for migration",
+        .mhandler.cmd = hmp_migrate_set_parameter,
+        .command_completion = migrate_set_parameter_completion,
+    },
+
+STEXI
+@item migrate_set_parameter @var{parameter} @var{value}
+@findex migrate_set_parameter
+Set the parameter @var{parameter} for migration.
+ETEXI
+
+    {
         .name       = "client_migrate_info",
         .args_type  = "protocol:s,hostname:s,port:i?,tls-port:i?,cert-subject:s?",
         .params     = "protocol hostname port tls-port cert-subject",
@@ -1764,6 +1779,8 @@ show user network stack connection states
 show migration status
 @item info migrate_capabilities
 show current migration capabilities
+@item info migrate_parameters
+show current migration parameters
 @item info migrate_cache_size
 show current migration XBZRLE cache size
 @item info balloon
diff --git a/hmp.c b/hmp.c
index b47f331..1f67651 100644
--- a/hmp.c
+++ b/hmp.c
@@ -246,6 +246,27 @@ void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict)
     qapi_free_MigrationCapabilityStatusList(caps);
 }
 
+void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
+{
+    MigrationParameterStatusList *params, *p;
+    MigrationParameterInt *data;
+
+    params = qmp_query_migrate_parameters(NULL);
+
+    if (params) {
+        monitor_printf(mon, "parameters:");
+        for (p = params; p; p = p->next) {
+            data = (MigrationParameterInt *)p->value->data;
+            monitor_printf(mon, " %s: %" PRId64,
+                           MigrationParameter_lookup[p->value->kind],
+                           data->value);
+        }
+        monitor_printf(mon, "\n");
+    }
+
+    qapi_free_MigrationParameterStatusList(params);
+}
+
 void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict)
 {
     monitor_printf(mon, "xbzrel cache size: %" PRId64 " kbytes\n",
@@ -1140,6 +1161,41 @@ void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict)
     }
 }
 
+void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
+{
+    const char *param = qdict_get_str(qdict, "parameter");
+    int value = qdict_get_int(qdict, "value");
+    Error *err = NULL;
+    MigrationParameterStatusList *params = g_malloc0(sizeof(*params));
+    MigrationParameterInt *data;
+    int i;
+
+    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+        if (strcmp(param, MigrationParameter_lookup[i]) == 0) {
+            params->value = g_malloc0(sizeof(*params->value));
+            params->value->kind = i;
+            params->value->data = g_malloc0(sizeof(MigrationParameterInt));
+            data = (MigrationParameterInt *)params->value->data;
+            data->value = value;
+            params->next = NULL;
+            qmp_migrate_set_parameters(params, &err);
+            break;
+        }
+    }
+
+    if (i == MIGRATION_PARAMETER_MAX) {
+        error_set(&err, QERR_INVALID_PARAMETER, param);
+    }
+
+    qapi_free_MigrationParameterStatusList(params);
+
+    if (err) {
+        monitor_printf(mon, "migrate_set_parameter: %s\n",
+                       error_get_pretty(err));
+        error_free(err);
+    }
+}
+
 void hmp_set_password(Monitor *mon, const QDict *qdict)
 {
     const char *protocol  = qdict_get_str(qdict, "protocol");
diff --git a/hmp.h b/hmp.h
index 4bb5dca..b2b2d2c 100644
--- a/hmp.h
+++ b/hmp.h
@@ -28,6 +28,7 @@ void hmp_info_chardev(Monitor *mon, const QDict *qdict);
 void hmp_info_mice(Monitor *mon, const QDict *qdict);
 void hmp_info_migrate(Monitor *mon, const QDict *qdict);
 void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict);
+void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict);
 void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict);
 void hmp_info_cpus(Monitor *mon, const QDict *qdict);
 void hmp_info_block(Monitor *mon, const QDict *qdict);
@@ -63,6 +64,7 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict);
+void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_cache_size(Monitor *mon, const QDict *qdict);
 void hmp_set_password(Monitor *mon, const QDict *qdict);
 void hmp_expire_password(Monitor *mon, const QDict *qdict);
@@ -111,6 +113,8 @@ void watchdog_action_completion(ReadLineState *rs, int nb_args,
                                 const char *str);
 void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
                                        const char *str);
+void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
+                                      const char *str);
 void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str);
 void host_net_remove_completion(ReadLineState *rs, int nb_args,
                                 const char *str);
diff --git a/include/migration/migration.h b/include/migration/migration.h
index 9ac1b23..434cc96 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -51,9 +51,7 @@ struct MigrationState
     QEMUBH *cleanup_bh;
     QEMUFile *file;
     QemuThread *compress_thread;
-    int compress_thread_count;
-    int decompress_thread_count;
-    int compress_level;
+    int parameters[MIGRATION_PARAMETER_MAX];
 
     int state;
     MigrationParams params;
diff --git a/migration/migration.c b/migration/migration.c
index 55f749e..89750ba 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -66,9 +66,12 @@ MigrationState *migrate_get_current(void)
         .bandwidth_limit = MAX_THROTTLE,
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
-        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
-        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
-        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
+        .parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] =
+                DEFAULT_MIGRATE_COMPRESS_LEVEL,
+        .parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
+                DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+                DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
     };
 
     return &current_migration;
@@ -178,6 +181,33 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
     return head;
 }
 
+MigrationParameterStatusList *qmp_query_migrate_parameters(Error **errp)
+{
+    MigrationParameterStatusList *head = NULL;
+    MigrationParameterStatusList *params;
+    MigrationState *s = migrate_get_current();
+    MigrationParameterInt *data;
+    int i;
+
+    params = NULL; /* silence compiler warning */
+    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+        if (head == NULL) {
+            head = g_malloc0(sizeof(*params));
+            params = head;
+        } else {
+            params->next = g_malloc0(sizeof(*params));
+            params = params->next;
+        }
+        params->value = g_malloc(sizeof(*params->value));
+        params->value->kind = i;
+        params->value->data = g_malloc(sizeof(MigrationParameterInt));
+        data = (MigrationParameterInt *)params->value->data;
+        data->value = s->parameters[i];
+    }
+
+    return head;
+}
+
 static void get_xbzrle_cache_stats(MigrationInfo *info)
 {
     if (migrate_use_xbzrle()) {
@@ -294,6 +324,44 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
     }
 }
 
+void qmp_migrate_set_parameters(MigrationParameterStatusList *params,
+                                Error **errp)
+{
+    MigrationState *s = migrate_get_current();
+    MigrationParameterStatusList *p;
+    MigrationParameterInt *data;
+
+    for (p = params; p; p = p->next) {
+        switch (p->value->kind) {
+        case MIGRATION_PARAMETER_COMPRESS_LEVEL:
+            data = (MigrationParameterInt *)p->value->data;
+            if (data->value < 0 || data->value > 9) {
+                error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
+                          "is invalid, it should be in the range of 0 to 9");
+                return;
+            }
+            break;
+        case MIGRATION_PARAMETER_COMPRESS_THREADS:
+        case MIGRATION_PARAMETER_DECOMPRESS_THREADS:
+            if (s->state == MIG_STATE_ACTIVE || s->state == MIG_STATE_SETUP) {
+                error_set(errp, QERR_MIGRATION_ACTIVE);
+                return;
+            }
+            data = (MigrationParameterInt *)p->value->data;
+            if (data->value < 1 || data->value > 255) {
+                error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                          "(de)compress_threads",
+                          "is invalid, it should be in the range of 1 to 255");
+                return;
+            }
+            break;
+        default:
+           return;
+        }
+        s->parameters[p->value->kind] = data->value;
+    }
+}
+
 /* shared migration helpers */
 
 static void migrate_set_state(MigrationState *s, int old_state, int new_state)
@@ -400,9 +468,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
     int64_t bandwidth_limit = s->bandwidth_limit;
     bool enabled_capabilities[MIGRATION_CAPABILITY_MAX];
     int64_t xbzrle_cache_size = s->xbzrle_cache_size;
-    int compress_level = s->compress_level;
-    int compress_thread_count = s->compress_thread_count;
-    int decompress_thread_count = s->decompress_thread_count;
+    int compress_level = s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
+    int compress_thread_count =
+            s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
+    int decompress_thread_count =
+            s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
 
     memcpy(enabled_capabilities, s->enabled_capabilities,
            sizeof(enabled_capabilities));
@@ -413,9 +483,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
            sizeof(enabled_capabilities));
     s->xbzrle_cache_size = xbzrle_cache_size;
 
-    s->compress_level = compress_level;
-    s->compress_thread_count = compress_thread_count;
-    s->decompress_thread_count = decompress_thread_count;
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
+               compress_thread_count;
+    s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+               decompress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -603,7 +675,7 @@ int migrate_compress_level(void)
 
     s = migrate_get_current();
 
-    return s->compress_level;
+    return s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
 }
 
 int migrate_compress_threads(void)
@@ -612,7 +684,7 @@ int migrate_compress_threads(void)
 
     s = migrate_get_current();
 
-    return s->compress_thread_count;
+    return s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
 }
 
 int migrate_decompress_threads(void)
@@ -621,7 +693,7 @@ int migrate_decompress_threads(void)
 
     s = migrate_get_current();
 
-    return s->decompress_thread_count;
+    return s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
 }
 
 int migrate_use_xbzrle(void)
diff --git a/monitor.c b/monitor.c
index c3cc060..499ae1c 100644
--- a/monitor.c
+++ b/monitor.c
@@ -2873,6 +2873,13 @@ static mon_cmd_t info_cmds[] = {
         .mhandler.cmd = hmp_info_migrate_capabilities,
     },
     {
+        .name       = "migrate_parameters",
+        .args_type  = "",
+        .params     = "",
+        .help       = "show current migration parameters",
+        .mhandler.cmd = hmp_info_migrate_parameters,
+    },
+    {
         .name       = "migrate_cache_size",
         .args_type  = "",
         .params     = "",
@@ -4555,6 +4562,24 @@ void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
     }
 }
 
+void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
+                                      const char *str)
+{
+    size_t len;
+
+    len = strlen(str);
+    readline_set_completion_index(rs, len);
+    if (nb_args == 2) {
+        int i;
+        for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+            const char *name = MigrationParameter_lookup[i];
+            if (!strncmp(str, name, len)) {
+                readline_add_completion(rs, name);
+            }
+        }
+    }
+}
+
 void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str)
 {
     int i;
diff --git a/qapi-schema.json b/qapi-schema.json
index 0dfc4ce..5bf21fe 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -541,6 +541,92 @@
 ##
 { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
 
+# @MigrationParameter
+#
+# Migration parameters enumeration
+#
+# @compress-level: Set the compression level to be used in live migration,
+#          the compression level is an integer between 0 and 9, where 0 means
+#          no compression, 1 means the best compression speed, and 9 means best
+#          compression ratio which will consume more CPU.
+#
+# @compress-threads: Set compression thread count to be used in live migration,
+#          the compression thread count is an integer between 1 and 255.
+#
+# @decompress-threads: Set decompression thread count to be used in live
+#          migration, the decompression thread count is an integer between 1
+#          and 255. Usually, decompression is at least 4 times as fast as
+#          compression, so set the decompress-threads to the number about 1/4
+#          of compress-threads is adequate.
+#
+# Since: 2.3
+##
+{ 'enum': 'MigrationParameter',
+  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
+##
+# @MigrationParameterBase
+#
+# Migration parameter information
+#
+# @parameter: the parameter of migration
+#
+# Since: 2.3
+##
+{ 'type': 'MigrationParameterBase',
+  'data': {'parameter': 'MigrationParameter'} }
+##
+# @MigrationParameterInt
+#
+# Migration parameter information
+#
+# @value: parameter int
+#
+# Since: 2.3
+##
+{ 'type': 'MigrationParameterInt',
+  'data': {'value': 'int'} }
+##
+# @MigrationParameterStatus
+#
+# Migration parameter information
+#
+# @compress-level: compression level
+#
+# @compress-threads: compression thread count
+#
+# @decompress-threads: decompression thread count
+#
+# Since: 2.3
+##
+{ 'union': 'MigrationParameterStatus',
+  'base': 'MigrationParameterBase',
+  'discriminator': 'parameter',
+  'data': { 'compress-level': 'MigrationParameterInt',
+            'compress-threads': 'MigrationParameterInt',
+            'decompress-threads': 'MigrationParameterInt'} }
+#
+# @migrate-set-parameters
+#
+# Set the following migration parameters (like compress-level)
+#
+# @parameters: json array of parameter modifications to make
+#
+# Since: 2.3
+##
+{ 'command': 'migrate-set-parameters',
+  'data': { 'parameters': ['MigrationParameterStatus'] } }
+##
+# @query-migrate-parameters
+#
+# Returns information about the current migration parameters status
+#
+# Returns: @MigrationParametersStatus
+#
+# Since: 2.3
+##
+{ 'command': 'query-migrate-parameters',
+  'returns': ['MigrationParameterStatus'] }
+##
 ##
 # @MouseInfo:
 #
diff --git a/qmp-commands.hx b/qmp-commands.hx
index a85d847..2c4737b 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -3295,6 +3295,72 @@ EQMP
     },
 
 SQMP
+migrate-set-parameters
+----------------------
+
+Set migration parameters
+
+- "compress-level": set compression level during migration
+- "compress-threads": set compression thread count for migration
+- "decompress-threads": set decompression thread count for migration
+
+Arguments:
+
+Example:
+
+-> { "execute": "migrate-set-parameters" , "arguments":
+     { "parameters": [ { "parameter": "compress-level", "value": 1 } ] } }
+
+EQMP
+
+    {
+        .name       = "migrate-set-parameters",
+        .args_type  = "parameters:O",
+        .params     = "parameter:s,value:O",
+	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
+    },
+SQMP
+query-migrate-parameters
+------------------------
+
+Query current migration parameters
+
+- "parameters": migration parameters value
+         - "compress-level" : compression level value (json-int)
+         - "compress-threads" : compression thread count value (json-int)
+         - "decompress-threads" : decompression thread count value (json-int)
+
+Arguments:
+
+Example:
+
+-> { "execute": "query-migrate-parameters" }
+<- {
+      "return": [
+         {
+            "parameter": "compress-level",
+            "value": 1
+         },
+         {
+            "parameter": "compress-threads",
+            "value": 8
+         },
+         {
+            "parameter": "decompress-threads",
+            "value": 2
+         }
+      ]
+   }
+
+EQMP
+
+    {
+        .name       = "query-migrate-parameters",
+        .args_type  = "",
+        .mhandler.cmd_new = qmp_marshal_input_query_migrate_parameters,
+    },
+
+SQMP
 query-balloon
 -------------
 
-- 
1.9.1

Re: [Qemu-devel] [v5 01/12] docs: Add a doc about multiple thread compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Give some details about the multiple thread (de)compression and
> how to use it in live migration.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

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

Re: [Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the code to create and destroy the multiple threads those will
> be used to do data compression. Left some functions empty to keep
> clearness, and the code will be added later.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

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


> +int migrate_compress_threads(void)
> +{
> +    MigrationState *s;
> +
> +    s = migrate_get_current();
> +
> +    return s->compress_thread_count;
> +}
> +

As far as I can see, all users have already s in place. No that I
disagree with an acessor function.

Re: [Qemu-devel] [v5 03/12] migration: Add the framework of multi-thread decompression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the code to create and destroy the multiple threads those will be
> used to do data decompression. Left some functions empty just to keep
> clearness, and the code will be added later.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

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

Re: [Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the code to create and destroy the multiple threads those will
> be used to do data compression. Left some functions empty to keep
> clearness, and the code will be added later.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

/me goes back

> +struct CompressParam {
> +    /* To be done */
> +};
> +typedef struct CompressParam CompressParam;
> +
> +static CompressParam *comp_param;
> +static bool quit_comp_thread;
> +
> +static void *do_data_compress(void *opaque)
> +{
> +    while (!quit_comp_thread) {

Using this variable without any protection, read from all threads.

> +
> +    /* To be done */
> +
> +    }
> +
> +    return NULL;
> +}
> +
> +static inline void terminate_compression_threads(void)
> +{
> +    quit_comp_thread = true;

written from the main migration thread.  Shouldn't we use a mutex or at
least atomic accessors?

PD. I haven't yet finishing reading the series, in case this is fixed later.

Re: [Qemu-devel] [v5 05/12] arch_init: Alloc and free data struct for compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Define the data structure and variables used to do multiple thread
> compression, and add the code to initialize and free them.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

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

Re: [Qemu-devel] [v5 06/12] arch_init: Add and free data struct for decompression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Define the data structure and variables used to do multiple thread
> decompression, and add the code to initialize and free them.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

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

Re: [Qemu-devel] [v5 07/12] migration: Split the function ram_save_page

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Split the function ram_save_page for code reuse purpose.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

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

Re: [Qemu-devel] [v5 07/12] migration: Split the function ram_save_page

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Split the function ram_save_page for code reuse purpose.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

you can do s/ram_save_page/ram_save_zero_page/ on $subject?

Re: [Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the code to create and destroy the multiple threads those will
> be used to do data compression. Left some functions empty to keep
> clearness, and the code will be added later.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

And here I am again.

Reviewing patch 8, I found that we need to fix some things here.

> +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
> +                                    ram_addr_t offset, bool last_stage)
> +{
> +    int bytes_sent = -1;
> +
> +    /* To be done*/
> +
> +    return bytes_sent;
> +}

We have three return values, here, that are not the same that for normal
pages

 0: this is the 1st page for a particular thread, nothing to sent yet
 n > 0: we are sending the previous compresed page for the choosen
        thread

Notice that the only way that ram_save_page() can return 0 is for xbzrle
when a page has modified but it has exactly the same value that before.

(it can have been modified twice, +1, -1 or whatever)

Notice that ram_save_page() can only return 0 (duplicate page) or > 0
(real size written)

> +
>  /*
>   * ram_find_and_save_block: Finds a page to send and sends it to f
>   *
> @@ -679,7 +751,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
>                  ram_bulk_stage = false;
>              }
>          } else {
> -            bytes_sent = ram_save_page(f, block, offset, last_stage);
> +            if (migrate_use_compression()) {
> +                bytes_sent = ram_save_compressed_page(f, block, offset,
> +                                                      last_stage);
> +            } else {
> +                bytes_sent = ram_save_page(f, block, offset, last_stage);
> +            }


I need more context, this is the corrent code

        } else {
            bytes_sent = ram_save_page(f, block, offset, last_stage);

            /* if page is unmodified, continue to the next */
            if (bytes_sent > 0) {
                last_sent_block = block;
                break;
            }
        }

And we should change to:

        } else if (migrate_use_compression()) {
            bytes_sent = ram_save_compressed_page(f, block, offset,
                                                  last_stage);
            last_sent_block = block;
            break;
        } else {
            bytes_sent = ram_save_page(f, block, offset, last_stage);

            /* if page is unmodified, continue to the next */
            if (bytes_sent > 0) {
                last_sent_block = block;
                break;
            }
        }

This would mean that we don't need to arrange for the zero byte return
on qemu.

Re: [Qemu-devel] [v5 08/12] migration: Add the core code of multi-thread compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Implement the core logic of the multiple thread compression. At this
> point, multiple thread compression can't co-work with xbzrle yet.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>


> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -363,18 +363,44 @@ static QemuMutex *comp_done_lock;
>  static QemuCond *comp_done_cond;
>  /* The empty QEMUFileOps will be used by file in CompressParam */
>  static const QEMUFileOps empty_ops = { };
> +
> +/* one_byte_count is used to count the bytes that is added to
> + * bytes_transferred but not actually transferred, at the proper
> + * time, we should sub one_byte_count from bytes_transferred to
> + * make bytes_transferred accurate.
> + */
> +static int one_byte_count;

With the changes proposed previously to ram_save_compressed_page() this
shouldn't be needed.  It can return 0 now.

> +static int do_compress_ram_page(CompressParam *param);
> +
>  static void *do_data_compress(void *opaque)
>  {
> -    while (!quit_comp_thread) {
> -
> -    /* To be done */
> +    CompressParam *param = opaque;
>  
> +    while (!quit_comp_thread) {
> +        qemu_mutex_lock(&param->mutex);
> +        /* Re-check the quit_comp_thread in case of
> +         * terminate_compression_threads is called just before
> +         * qemu_mutex_lock(&param->mutex) and after
> +         * while(!quit_comp_thread), re-check it here can make
> +         * sure the compression thread terminate as expected.
> +         */
> +        while (!param->busy && !quit_comp_thread) {
> +            qemu_cond_wait(&param->cond, &param->mutex);
> +        }
> +        qemu_mutex_unlock(&param->mutex);
> +        if (!quit_comp_thread) {
> +            do_compress_ram_page(param);
> +        }
> +        qemu_mutex_lock(comp_done_lock);
> +        param->busy = false;
> +        qemu_cond_signal(comp_done_cond);
> +        qemu_mutex_unlock(comp_done_lock);
>      }
>  
>      return NULL;
> @@ -382,9 +408,15 @@ static void *do_data_compress(void *opaque)
>  
>  static inline void terminate_compression_threads(void)
>  {
> -    quit_comp_thread = true;
> +    int idx, thread_count;
>  
> -    /* To be done */
> +    thread_count = migrate_compress_threads();
> +    quit_comp_thread = true;
> +    for (idx = 0; idx < thread_count; idx++) {
> +        qemu_mutex_lock(&comp_param[idx].mutex);
> +        qemu_cond_signal(&comp_param[idx].cond);
> +        qemu_mutex_unlock(&comp_param[idx].mutex);
> +    }
>  }
>  
>  void migrate_compress_threads_join(MigrationState *s)
> @@ -770,12 +802,157 @@ static int ram_save_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset,
>      return bytes_sent;
>  }
>  
> +static int do_compress_ram_page(CompressParam *param)
> +{
> +    int bytes_sent, cont;
> +    int blen;
> +    uint8_t *p;
> +    RAMBlock *block = param->block;
> +    ram_addr_t offset = param->offset;
> +
> +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> +    p = memory_region_get_ram_ptr(block->mr) + offset;
> +
> +    bytes_sent = save_block_hdr(param->file, block, offset, cont,
> +                                RAM_SAVE_FLAG_COMPRESS_PAGE);
> +    blen = qemu_put_compression_data(param->file, p, TARGET_PAGE_SIZE,
> +                                     migrate_compress_level());
> +    bytes_sent += blen;
> +    atomic_inc(&acct_info.norm_pages);
> +
> +    return bytes_sent;
> +}
> +
> +static inline void start_compression(CompressParam *param)
> +{
> +    qemu_mutex_lock(&param->mutex);
> +    param->busy = true;
> +    qemu_cond_signal(&param->cond);
> +    qemu_mutex_unlock(&param->mutex);
> +}
> +
> +
> +static uint64_t bytes_transferred;
> +
> +static void flush_compressed_data(QEMUFile *f)
> +{
> +    int idx, len, thread_count;
> +
> +    if (!migrate_use_compression()) {
> +        return;
> +    }
> +    thread_count = migrate_compress_threads();
> +    for (idx = 0; idx < thread_count; idx++) {
> +        if (comp_param[idx].busy) {
> +            qemu_mutex_lock(comp_done_lock);
> +            while (comp_param[idx].busy && !quit_comp_thread) {
> +                qemu_cond_wait(comp_done_cond, comp_done_lock);
> +            }
> +            qemu_mutex_unlock(comp_done_lock);
> +        }

If we arrive here because quit_comp_thread == true, shouldn't we skip
the qemu_put_qemu_file()?

> +        len = qemu_put_qemu_file(f, comp_param[idx].file);
> +        bytes_transferred += len;
> +    }

[remove one_byte stuff here]

> +}
> +
> +static inline void set_compress_params(CompressParam *param, RAMBlock *block,
> +                                       ram_addr_t offset)
> +{
> +    param->block = block;
> +    param->offset = offset;
> +}
> +
> +static int compress_page_with_multi_thread(QEMUFile *f, RAMBlock *block,
> +                                           ram_addr_t offset)
> +{
> +    int idx, thread_count, bytes_sent = 0;
> +
> +    thread_count = migrate_compress_threads();
> +    qemu_mutex_lock(comp_done_lock);
> +    while (true) {
> +        for (idx = 0; idx < thread_count; idx++) {
> +            if (!comp_param[idx].busy) {
> +                bytes_sent = qemu_put_qemu_file(f, comp_param[idx].file);
> +                set_compress_params(&comp_param[idx], block, offset);
> +                start_compression(&comp_param[idx]);

[remove stuff here]

> +                break;
> +            }
> +        }
> +        if (bytes_sent > 0) {

Change this to:
          if (bytes_sent >= 0) {

> +            break;
> +        } else {
> +            qemu_cond_wait(comp_done_cond, comp_done_lock);
> +        }
> +    }
> +    qemu_mutex_unlock(comp_done_lock);
> +
> +    return bytes_sent;
> +}
> +
>  static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
>                                      ram_addr_t offset, bool last_stage)
>  {
>      int bytes_sent = -1;
> +    MemoryRegion *mr = block->mr;
> +    uint8_t *p;
> +    int ret;
> +    int cont;
>  
> -    /* To be done*/
> +    p = memory_region_get_ram_ptr(mr) + offset;
> +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> +    ret = ram_control_save_page(f, block->offset,
> +                                offset, TARGET_PAGE_SIZE, &bytes_sent);
> +    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
> +        if (ret != RAM_SAVE_CONTROL_DELAYED) {
> +            if (bytes_sent > 0) {
> +                acct_info.norm_pages++;
> +            } else if (bytes_sent == 0) {
> +                acct_info.dup_pages++;
> +            }
> +        }
> +    } else {
> +        /* When starting the process of a new block, the first page of
> +         * the block should be sent out before other pages in the same
> +         * block, and all the pages in last block should have been sent
> +         * out, keeping this order is important, because the 'cont' flag
> +         * is used to avoid resending the block name.
> +         */
> +        if (block != last_sent_block) {
> +            flush_compressed_data(f);
> +            bytes_sent = save_zero_page(f, block, offset, p, cont);
> +            if (bytes_sent == -1) {
> +                set_compress_params(&comp_param[0], block, offset);
> +                /* Use the qemu thread to compress the data to make sure the
> +                 * first page is sent out before other pages
> +                 */
> +                bytes_sent = do_compress_ram_page(&comp_param[0]);
> +                if (bytes_sent > 0) {

This test is not needed

assert(bytes_sent>0)

or how can it be zero or negative here?  So, we have to always call
qemu_put_qemu_file() no?

> +                    qemu_put_qemu_file(f, comp_param[0].file);
> +                }
> +            }
> +        } else {
> +            bytes_sent = save_zero_page(f, block, offset, p, cont);
> +            if (bytes_sent == -1) {
> +                bytes_sent = compress_page_with_multi_thread(f, block, offset);
> +            }
> +        }
> +    }
>  
>      return bytes_sent;
>  }
> @@ -834,8 +1011,6 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
>      return bytes_sent;
>  }
>  
> -static uint64_t bytes_transferred;
> -
>  void acct_update_position(QEMUFile *f, size_t size, bool zero)
>  {
>      uint64_t pages = size / TARGET_PAGE_SIZE;
> @@ -1043,6 +1218,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
>          i++;
>      }
>  
> +    flush_compressed_data(f);
>      qemu_mutex_unlock_ramlist();
>  
>      /*
> @@ -1089,6 +1265,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
>          bytes_transferred += bytes_sent;
>      }
>  
> +    flush_compressed_data(f);
>      ram_control_after_iterate(f, RAM_CONTROL_FINISH);
>      migration_end();


I thihnk this would make the code work, but not the locking.  You are
using here:

quit_comp_thread:  global, and not completely clear what protects it
comp_done_lock: global
comp_done_cond: global

param[i].busy: I would suggest renaming to pending work
param[i].mutex:
param[i].cond:
       thread is waiting for work


Issues:

param->busy is protected on do_data_compress() and start_compression()
with param->busy, but in flush_compressed_data() and
comress_page_with_multithread() it is protected by
comp_done_lock.

At this point, I would suggest to just drop param[i].mutex and use
everywhere comp_done_lock.  We can make locking granularly later if
needed, but 1st get it correct?

Code basically does (forget termination and locking)

each compression_thread()

  while(1) {
     while(!work_to_do)
        wait_for_work
     do_work
  }

And the main thread does:


while(1) {
     foreacth compression_thread {
          if thread free {
             put it to work
             break;
          }
          wait_for_thread_to_finish
     }
}

Notice how we are walking all threads each time that we need to do anything

Perhaps code should be more simple if we put the data that needs to be
done on a global variable and change this to:

compression_thread

  while(1) {
     while(!work_to_do)
        wait_for_work
     pick work from global variable
     wakeup main thread
     do_work
  }

main thread:

put work on global variable
while(nobody_pick_thework) {
     signal all threads
     wait for a compression thread to take the work
}

Why?  because then we only have a global mutex and two condition
variables, with a clear semantics:
- lock protects two conditions and global variable with work
- one condition where threads wait for work
- one condition where main thread wait for a worker to be ready

As we would need to lock every single tame to put the work in the global
variable, to wait or to pick up the work, we can stop all the:

if (!foo) {
    mutex_lock
    if(!foo) /* this time with lock */
        ....
}


Sorry for the very long mail, if it makes you feel better, this is the
second time that I wrote it, because the 1st version, my locking
proposal didn't worked correctly.

What do you think?

Later, Juan.

Re: [Qemu-devel] [v5 09/12] migration: Make compression co-work with xbzrle

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Now, multiple thread compression can co-work with xbzrle. when
> xbzrle is on, multiple thread compression will only work at the
> first round of RAM data sync.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>

Drop this patch and just give an error when trying to set xbzrle and
compression?  User have to pick one and only one, no second guess him/her?

Later, Juan.

Re: [Qemu-devel] [v5 10/12] migration: Add the core code for decompression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Implement the core logic of multiple thread decompression,
> the decompression can work now.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

I think it should be a good idea to use the same scheme for compression
and decompression.  So, if you decide to use my proposal for locking
there, please do the same here.

Later, Juan.

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the qmp and hmp commands to tune and query the parameters used in
> live migration.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

I agree with the existence of this changes, and we could use them for
existing paramenters like migration_speed (independent of this patch).

About the qmp bits of the patch, I let others to comment on it.

I like the appreach, though.

Later, Juan.

Re: [Qemu-devel] [v5 09/12] migration: Make compression co-work with xbzrle

[Li, Liang Z]

> Liang Li <liang.z.li@intel.com> wrote:
> > Now, multiple thread compression can co-work with xbzrle. when xbzrle
> > is on, multiple thread compression will only work at the first round
> > of RAM data sync.
> >
> > Signed-off-by: Liang Li <liang.z.li@intel.com>
> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
> 
> Drop this patch and just give an error when trying to set xbzrle and
> compression?  User have to pick one and only one, no second guess him/her?
> 

Live migration can benefit from compression co-work with xbzrle. You know, xbzrle 
transfer the raw RAM pages to destination in the ram bulk stage, and after that, it transfers
the diff data. The ram bulk stage is where compression can do optimization, and beside 
the ram bulk stage, xbzrle may do better than compression  in some situation. So
compression and xbzrle are not in conflict but complementary.

I think it's a pity if we limit the use to select only one of them. If there is no strong reason, 
I don't agree to drop this patch.

Liang

Re: [Qemu-devel] [v5 05/12] arch_init: Alloc and free data struct for compression

[Li, Liang Z]

Hi Juan,

     Have you reviewed the 04 patch of the patch series? I didn't see the reply email.

Liang

> -----Original Message-----
> From: Juan Quintela [mailto:quintela@redhat.com]
> Sent: Wednesday, February 11, 2015 5:03 PM
> To: Li, Liang Z
> Cc: qemu-devel@nongnu.org; eblake@redhat.com; amit.shah@redhat.com;
> lcapitulino@redhat.com; armbru@redhat.com; dgilbert@redhat.com; Zhang,
> Yang Z
> Subject: Re: [v5 05/12] arch_init: Alloc and free data struct for compression
> 
> Liang Li <liang.z.li@intel.com> wrote:
> > Define the data structure and variables used to do multiple thread
> > compression, and add the code to initialize and free them.
> >
> > Signed-off-by: Liang Li <liang.z.li@intel.com>
> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
> 
> Reviewed-by: Juan Quintela <quintela@redhat.com>

Re: [Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Li, Liang Z]

> Reviewing patch 8, I found that we need to fix some things here.
> 
> > +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
> > +                                    ram_addr_t offset, bool
> > +last_stage) {
> > +    int bytes_sent = -1;
> > +
> > +    /* To be done*/
> > +
> > +    return bytes_sent;
> > +}
> 
> We have three return values, here, that are not the same that for normal
> pages
> 
>  0: this is the 1st page for a particular thread, nothing to sent yet  n > 0: we
> are sending the previous compresed page for the choosen
>         thread
> 
> Notice that the only way that ram_save_page() can return 0 is for xbzrle
> when a page has modified but it has exactly the same value that before.
> 
> (it can have been modified twice, +1, -1 or whatever)
> 
> Notice that ram_save_page() can only return 0 (duplicate page) or > 0 (real
> size written)
> 
> > +
> >  /*
> >   * ram_find_and_save_block: Finds a page to send and sends it to f
> >   *
> > @@ -679,7 +751,12 @@ static int ram_find_and_save_block(QEMUFile *f,
> bool last_stage)
> >                  ram_bulk_stage = false;
> >              }
> >          } else {
> > -            bytes_sent = ram_save_page(f, block, offset, last_stage);
> > +            if (migrate_use_compression()) {
> > +                bytes_sent = ram_save_compressed_page(f, block, offset,
> > +                                                      last_stage);
> > +            } else {
> > +                bytes_sent = ram_save_page(f, block, offset, last_stage);
> > +            }
> 
> 
> I need more context, this is the corrent code
> 
>         } else {
>             bytes_sent = ram_save_page(f, block, offset, last_stage);
> 
>             /* if page is unmodified, continue to the next */
>             if (bytes_sent > 0) {
>                 last_sent_block = block;
>                 break;
>             }
>         }
> 
> And we should change to:
> 
>         } else if (migrate_use_compression()) {
>             bytes_sent = ram_save_compressed_page(f, block, offset,
>                                                   last_stage);
>             last_sent_block = block;
>             break;


What happened if ram_save_compressed_page() return 0 ?  following the flush_compressed_data() will be call,
The code call still work, but the efficiency is poor. Every time the main thread find there is no free compression
thread, it has to wait all compression threads finish their job before it can start the next round. 
The effective way is to start compression once there is any free compression thread.

Liang

>         } else {
>             bytes_sent = ram_save_page(f, block, offset, last_stage);
> 
>             /* if page is unmodified, continue to the next */
>             if (bytes_sent > 0) {
>                 last_sent_block = block;
>                 break;
>             }
>         }
> 
> This would mean that we don't need to arrange for the zero byte return on
> qemu.
> 

Re: [Qemu-devel] [v5 08/12] migration: Add the core code of multi-thread compression

[Li, Liang Z]

> -----Original Message-----
> From: Juan Quintela [mailto:quintela@redhat.com]
> Sent: Wednesday, February 11, 2015 7:45 PM
> To: Li, Liang Z
> Cc: qemu-devel@nongnu.org; eblake@redhat.com; amit.shah@redhat.com;
> lcapitulino@redhat.com; armbru@redhat.com; dgilbert@redhat.com; Zhang,
> Yang Z
> Subject: Re: [v5 08/12] migration: Add the core code of multi-thread
> compression
> 
> Liang Li <liang.z.li@intel.com> wrote:
> > Implement the core logic of the multiple thread compression. At this
> > point, multiple thread compression can't co-work with xbzrle yet.
> >
> > Signed-off-by: Liang Li <liang.z.li@intel.com>
> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> 
> 
> > --- a/arch_init.c
> > +++ b/arch_init.c
> > @@ -363,18 +363,44 @@ static QemuMutex *comp_done_lock;  static
> > QemuCond *comp_done_cond;
> >  /* The empty QEMUFileOps will be used by file in CompressParam */
> > static const QEMUFileOps empty_ops = { };
> > +
> > +/* one_byte_count is used to count the bytes that is added to
> > + * bytes_transferred but not actually transferred, at the proper
> > + * time, we should sub one_byte_count from bytes_transferred to
> > + * make bytes_transferred accurate.
> > + */
> > +static int one_byte_count;
> 
> With the changes proposed previously to ram_save_compressed_page() this
> shouldn't be needed.  It can return 0 now.
> 
> > +static int do_compress_ram_page(CompressParam *param);
> > +
> >  static void *do_data_compress(void *opaque)  {
> > -    while (!quit_comp_thread) {
> > -
> > -    /* To be done */
> > +    CompressParam *param = opaque;
> >
> > +    while (!quit_comp_thread) {
> > +        qemu_mutex_lock(&param->mutex);
> > +        /* Re-check the quit_comp_thread in case of
> > +         * terminate_compression_threads is called just before
> > +         * qemu_mutex_lock(&param->mutex) and after
> > +         * while(!quit_comp_thread), re-check it here can make
> > +         * sure the compression thread terminate as expected.
> > +         */
> > +        while (!param->busy && !quit_comp_thread) {
> > +            qemu_cond_wait(&param->cond, &param->mutex);
> > +        }
> > +        qemu_mutex_unlock(&param->mutex);
> > +        if (!quit_comp_thread) {
> > +            do_compress_ram_page(param);
> > +        }
> > +        qemu_mutex_lock(comp_done_lock);
> > +        param->busy = false;
> > +        qemu_cond_signal(comp_done_cond);
> > +        qemu_mutex_unlock(comp_done_lock);
> >      }
> >
> >      return NULL;
> > @@ -382,9 +408,15 @@ static void *do_data_compress(void *opaque)
> >
> >  static inline void terminate_compression_threads(void)
> >  {
> > -    quit_comp_thread = true;
> > +    int idx, thread_count;
> >
> > -    /* To be done */
> > +    thread_count = migrate_compress_threads();
> > +    quit_comp_thread = true;
> > +    for (idx = 0; idx < thread_count; idx++) {
> > +        qemu_mutex_lock(&comp_param[idx].mutex);
> > +        qemu_cond_signal(&comp_param[idx].cond);
> > +        qemu_mutex_unlock(&comp_param[idx].mutex);
> > +    }
> >  }
> >
> >  void migrate_compress_threads_join(MigrationState *s) @@ -770,12
> > +802,157 @@ static int ram_save_page(QEMUFile *f, RAMBlock *block,
> ram_addr_t offset,
> >      return bytes_sent;
> >  }
> >
> > +static int do_compress_ram_page(CompressParam *param) {
> > +    int bytes_sent, cont;
> > +    int blen;
> > +    uint8_t *p;
> > +    RAMBlock *block = param->block;
> > +    ram_addr_t offset = param->offset;
> > +
> > +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> > +    p = memory_region_get_ram_ptr(block->mr) + offset;
> > +
> > +    bytes_sent = save_block_hdr(param->file, block, offset, cont,
> > +                                RAM_SAVE_FLAG_COMPRESS_PAGE);
> > +    blen = qemu_put_compression_data(param->file, p,
> TARGET_PAGE_SIZE,
> > +                                     migrate_compress_level());
> > +    bytes_sent += blen;
> > +    atomic_inc(&acct_info.norm_pages);
> > +
> > +    return bytes_sent;
> > +}
> > +
> > +static inline void start_compression(CompressParam *param) {
> > +    qemu_mutex_lock(&param->mutex);
> > +    param->busy = true;
> > +    qemu_cond_signal(&param->cond);
> > +    qemu_mutex_unlock(&param->mutex); }
> > +
> > +
> > +static uint64_t bytes_transferred;
> > +
> > +static void flush_compressed_data(QEMUFile *f) {
> > +    int idx, len, thread_count;
> > +
> > +    if (!migrate_use_compression()) {
> > +        return;
> > +    }
> > +    thread_count = migrate_compress_threads();
> > +    for (idx = 0; idx < thread_count; idx++) {
> > +        if (comp_param[idx].busy) {
> > +            qemu_mutex_lock(comp_done_lock);
> > +            while (comp_param[idx].busy && !quit_comp_thread) {
> > +                qemu_cond_wait(comp_done_cond, comp_done_lock);
> > +            }
> > +            qemu_mutex_unlock(comp_done_lock);
> > +        }
> 
> If we arrive here because quit_comp_thread == true, shouldn't we skip the
> qemu_put_qemu_file()?
> 
> > +        len = qemu_put_qemu_file(f, comp_param[idx].file);
> > +        bytes_transferred += len;
> > +    }
> 
> [remove one_byte stuff here]
> 
> > +}
> > +
> > +static inline void set_compress_params(CompressParam *param,
> RAMBlock *block,
> > +                                       ram_addr_t offset) {
> > +    param->block = block;
> > +    param->offset = offset;
> > +}
> > +
> > +static int compress_page_with_multi_thread(QEMUFile *f, RAMBlock
> *block,
> > +                                           ram_addr_t offset) {
> > +    int idx, thread_count, bytes_sent = 0;
> > +
> > +    thread_count = migrate_compress_threads();
> > +    qemu_mutex_lock(comp_done_lock);
> > +    while (true) {
> > +        for (idx = 0; idx < thread_count; idx++) {
> > +            if (!comp_param[idx].busy) {
> > +                bytes_sent = qemu_put_qemu_file(f, comp_param[idx].file);
> > +                set_compress_params(&comp_param[idx], block, offset);
> > +                start_compression(&comp_param[idx]);
> 
> [remove stuff here]
> 
> > +                break;
> > +            }
> > +        }
> > +        if (bytes_sent > 0) {
> 
> Change this to:
>           if (bytes_sent >= 0) {
> 
> > +            break;
> > +        } else {
> > +            qemu_cond_wait(comp_done_cond, comp_done_lock);
> > +        }
> > +    }
> > +    qemu_mutex_unlock(comp_done_lock);
> > +
> > +    return bytes_sent;
> > +}
> > +
> >  static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
> >                                      ram_addr_t offset, bool
> > last_stage)  {
> >      int bytes_sent = -1;
> > +    MemoryRegion *mr = block->mr;
> > +    uint8_t *p;
> > +    int ret;
> > +    int cont;
> >
> > -    /* To be done*/
> > +    p = memory_region_get_ram_ptr(mr) + offset;
> > +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> > +    ret = ram_control_save_page(f, block->offset,
> > +                                offset, TARGET_PAGE_SIZE, &bytes_sent);
> > +    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
> > +        if (ret != RAM_SAVE_CONTROL_DELAYED) {
> > +            if (bytes_sent > 0) {
> > +                acct_info.norm_pages++;
> > +            } else if (bytes_sent == 0) {
> > +                acct_info.dup_pages++;
> > +            }
> > +        }
> > +    } else {
> > +        /* When starting the process of a new block, the first page of
> > +         * the block should be sent out before other pages in the same
> > +         * block, and all the pages in last block should have been sent
> > +         * out, keeping this order is important, because the 'cont' flag
> > +         * is used to avoid resending the block name.
> > +         */
> > +        if (block != last_sent_block) {
> > +            flush_compressed_data(f);
> > +            bytes_sent = save_zero_page(f, block, offset, p, cont);
> > +            if (bytes_sent == -1) {
> > +                set_compress_params(&comp_param[0], block, offset);
> > +                /* Use the qemu thread to compress the data to make sure the
> > +                 * first page is sent out before other pages
> > +                 */
> > +                bytes_sent = do_compress_ram_page(&comp_param[0]);
> > +                if (bytes_sent > 0) {
> 
> This test is not needed
> 
> assert(bytes_sent>0)
> 
> or how can it be zero or negative here?  So, we have to always call
> qemu_put_qemu_file() no?
> 
> > +                    qemu_put_qemu_file(f, comp_param[0].file);
> > +                }
> > +            }
> > +        } else {
> > +            bytes_sent = save_zero_page(f, block, offset, p, cont);
> > +            if (bytes_sent == -1) {
> > +                bytes_sent = compress_page_with_multi_thread(f, block, offset);
> > +            }
> > +        }
> > +    }
> >
> >      return bytes_sent;
> >  }
> > @@ -834,8 +1011,6 @@ static int ram_find_and_save_block(QEMUFile *f,
> bool last_stage)
> >      return bytes_sent;
> >  }
> >
> > -static uint64_t bytes_transferred;
> > -
> >  void acct_update_position(QEMUFile *f, size_t size, bool zero)  {
> >      uint64_t pages = size / TARGET_PAGE_SIZE; @@ -1043,6 +1218,7 @@
> > static int ram_save_iterate(QEMUFile *f, void *opaque)
> >          i++;
> >      }
> >
> > +    flush_compressed_data(f);
> >      qemu_mutex_unlock_ramlist();
> >
> >      /*
> > @@ -1089,6 +1265,7 @@ static int ram_save_complete(QEMUFile *f, void
> *opaque)
> >          bytes_transferred += bytes_sent;
> >      }
> >
> > +    flush_compressed_data(f);
> >      ram_control_after_iterate(f, RAM_CONTROL_FINISH);
> >      migration_end();
> 
> 
> I thihnk this would make the code work, but not the locking.  You are using
> here:
> 
> quit_comp_thread:  global, and not completely clear what protects it
> comp_done_lock: global
> comp_done_cond: global
> 
> param[i].busy: I would suggest renaming to pending work
> param[i].mutex:
> param[i].cond:
>        thread is waiting for work
> 
> 
> Issues:
> 
> param->busy is protected on do_data_compress() and start_compression()
> with param->busy, but in flush_compressed_data() and
> comress_page_with_multithread() it is protected by comp_done_lock.
> 
> At this point, I would suggest to just drop param[i].mutex and use
> everywhere comp_done_lock.  We can make locking granularly later if
> needed, but 1st get it correct?
> 
> Code basically does (forget termination and locking)
> 
> each compression_thread()
> 
>   while(1) {
>      while(!work_to_do)
>         wait_for_work
>      do_work
>   }
> 
> And the main thread does:
> 
> 
> while(1) {
>      foreacth compression_thread {
>           if thread free {
>              put it to work
>              break;
>           }
>           wait_for_thread_to_finish
>      }
> }
> 
> Notice how we are walking all threads each time that we need to do anything
> 
> Perhaps code should be more simple if we put the data that needs to be
> done on a global variable and change this to:
> 
> compression_thread
> 
>   while(1) {
>      while(!work_to_do)
>         wait_for_work
>      pick work from global variable
>      wakeup main thread
>      do_work
>   }
> 
> main thread:
> 
> put work on global variable
> while(nobody_pick_thework) {
>      signal all threads
>      wait for a compression thread to take the work }
> 
> Why?  because then we only have a global mutex and two condition variables,
> with a clear semantics:
> - lock protects two conditions and global variable with work
> - one condition where threads wait for work
> - one condition where main thread wait for a worker to be ready
> 
> As we would need to lock every single tame to put the work in the global
> variable, to wait or to pick up the work, we can stop all the:
> 
> if (!foo) {
>     mutex_lock
>     if(!foo) /* this time with lock */
>         ....
> }
> 
> 
> Sorry for the very long mail, if it makes you feel better, this is the second
> time that I wrote it, because the 1st version, my locking proposal didn't
> worked correctly.
> 
> What do you think?

It sounds good, I will try according to your suggestion.  Thanks for your detail explanation :)

Liang

Re: [Qemu-devel] [v5 05/12] arch_init: Alloc and free data struct for compression

[Juan Quintela]

"Li, Liang Z" <liang.z.li@intel.com> wrote:
> Hi Juan,
>
>      Have you reviewed the 04 patch of the patch series? I didn't see
> the reply email.

It is ok, going to put the review-by

>
> Liang
>
>> -----Original Message-----
>> From: Juan Quintela [mailto:quintela@redhat.com]
>> Sent: Wednesday, February 11, 2015 5:03 PM
>> To: Li, Liang Z
>> Cc: qemu-devel@nongnu.org; eblake@redhat.com; amit.shah@redhat.com;
>> lcapitulino@redhat.com; armbru@redhat.com; dgilbert@redhat.com; Zhang,
>> Yang Z
>> Subject: Re: [v5 05/12] arch_init: Alloc and free data struct for compression
>> 
>> Liang Li <liang.z.li@intel.com> wrote:
>> > Define the data structure and variables used to do multiple thread
>> > compression, and add the code to initialize and free them.
>> >
>> > Signed-off-by: Liang Li <liang.z.li@intel.com>
>> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
>> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
>> 
>> Reviewed-by: Juan Quintela <quintela@redhat.com>

Re: [Qemu-devel] [v5 04/12] qemu-file: Add compression functions to QEMUFile

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> qemu_put_compression_data() compress the data and put it to QEMUFile.
> qemu_put_qemu_file() put the data in the buffer of source QEMUFile to
> destination QEMUFile.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

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

Re: [Qemu-devel] [v5 09/12] migration: Make compression co-work with xbzrle

[Juan Quintela]

"Li, Liang Z" <liang.z.li@intel.com> wrote:
>> Liang Li <liang.z.li@intel.com> wrote:
>> > Now, multiple thread compression can co-work with xbzrle. when xbzrle
>> > is on, multiple thread compression will only work at the first round
>> > of RAM data sync.
>> >
>> > Signed-off-by: Liang Li <liang.z.li@intel.com>
>> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
>> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com>
>> 
>> Drop this patch and just give an error when trying to set xbzrle and
>> compression?  User have to pick one and only one, no second guess him/her?
>> 
>
> Live migration can benefit from compression co-work with xbzrle. You know, xbzrle 
> transfer the raw RAM pages to destination in the ram bulk stage, and
> after that, it transfers
> the diff data.

I don't have numbers, so it is just hand-waving for my part.

> The ram bulk stage is where compression can do
> optimization,

If we do compression in the bulk page, xbzrle cache is going to be empty
after that, so we need to re-send the whole page anyways (at least the
1st time).

> and beside
> the ram bulk stage, xbzrle may do better than compression  in some
> situation.

With your patch, there is no way to select xbzrle for bulk stage, and
compression for the rest.  So this is is not an argument for this patch O:-)

> So
> compression and xbzrle are not in conflict but complementary.

Oh, there is no conflict between them, I fully agree there.  The problem
is when to use one or when to use the other.

My proposal:
- user devices if it wants to use xbzrle or compression.  It is
  completely clear what is going to happen.

- with this patch:
  ram bulk stage use compressión and from there it uses xbzrle: this
  needs at least to be documented on the man page and command line,
  otherwise the user don't know.

- perhaps it is even a good idea to change the code to do

    if (is_zero_range(..))
        send_it_as_one_byte()
    if (it_is_on_bzrle_cache())
        send as bzrle()
    if (it is not on xbzrle_cache() {
        put it on xbzrle_cache()
        send it compressed()
    }

And I am sure that there are even more posibilities.  The problem is
that to choose one or another we should:
- meassure what is better
- decide what to implement
- document how it works

As far as I can see, here we are doing the second item without having
doing the 1st (or at least I haven't seen the results), and clearly
without doing the third.

> I think it's a pity if we limit the use to select only one of them. If
> there is no strong reason,
> I don't agree to drop this patch.

Then you need to at least add documentation to explain why/what you are
doing.  If user selectcts xbzrle, it is clear what it does.  If user
selects compress, it is also clear.  If it selects both, it is not clear
(for looking at the documentation).

Later, Juan.

Re: [Qemu-devel] [v5 02/12] migration: Add the framework of multi-thread compression

[Juan Quintela]

"Li, Liang Z" <liang.z.li@intel.com> wrote:
>> Reviewing patch 8, I found that we need to fix some things here.
>> 
>> > +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
>> > +                                    ram_addr_t offset, bool
>> > +last_stage) {
>> > +    int bytes_sent = -1;
>> > +
>> > +    /* To be done*/
>> > +
>> > +    return bytes_sent;
>> > +}
>> 
>> We have three return values, here, that are not the same that for normal
>> pages
>> 
>>  0: this is the 1st page for a particular thread, nothing to sent yet  n > 0: we
>> are sending the previous compresed page for the choosen
>>         thread
>> 
>> Notice that the only way that ram_save_page() can return 0 is for xbzrle
>> when a page has modified but it has exactly the same value that before.
>> 
>> (it can have been modified twice, +1, -1 or whatever)
>> 
>> Notice that ram_save_page() can only return 0 (duplicate page) or > 0 (real
>> size written)
>> 
>> > +
>> >  /*
>> >   * ram_find_and_save_block: Finds a page to send and sends it to f
>> >   *
>> > @@ -679,7 +751,12 @@ static int ram_find_and_save_block(QEMUFile *f,
>> bool last_stage)
>> >                  ram_bulk_stage = false;
>> >              }
>> >          } else {
>> > -            bytes_sent = ram_save_page(f, block, offset, last_stage);
>> > +            if (migrate_use_compression()) {
>> > +                bytes_sent = ram_save_compressed_page(f, block, offset,
>> > +                                                      last_stage);
>> > +            } else {
>> > +                bytes_sent = ram_save_page(f, block, offset, last_stage);
>> > +            }
>> 
>> 
>> I need more context, this is the corrent code
>> 
>>         } else {
>>             bytes_sent = ram_save_page(f, block, offset, last_stage);
>> 
>>             /* if page is unmodified, continue to the next */
>>             if (bytes_sent > 0) {
>>                 last_sent_block = block;
>>                 break;
>>             }
>>         }
>> 
>> And we should change to:
>> 
>>         } else if (migrate_use_compression()) {
>>             bytes_sent = ram_save_compressed_page(f, block, offset,
>>                                                   last_stage);
>>             last_sent_block = block;
>>             break;
>
>
> What happened if ram_save_compressed_page() return 0 ?  following the
> flush_compressed_data() will be call,

This happens to me to send suggestions instead of proper code.
You are right.

I fixed one of the callers, but not the "upstream" caller.


> The code call still work, but the efficiency is poor. Every time the
> main thread find there is no free compression
> thread, it has to wait all compression threads finish their job before
> it can start the next round.
> The effective way is to start compression once there is any free
> compression thread.


Will send to the list a suggestion to improve from here, right?

Sorry for the noise, Juan.

> Liang
>
>>         } else {
>>             bytes_sent = ram_save_page(f, block, offset, last_stage);
>> 
>>             /* if page is unmodified, continue to the next */
>>             if (bytes_sent > 0) {
>>                 last_sent_block = block;
>>                 break;
>>             }
>>         }
>> 
>> This would mean that we don't need to arrange for the zero byte return on
>> qemu.
>> 

Re: [Qemu-devel] [v5 09/12] migration: Make compression co-work with xbzrle

[Li, Liang Z]

> >> Drop this patch and just give an error when trying to set xbzrle and
> >> compression?  User have to pick one and only one, no second guess
> him/her?
> >>
> >
> > Live migration can benefit from compression co-work with xbzrle. You
> > know, xbzrle transfer the raw RAM pages to destination in the ram bulk
> > stage, and after that, it transfers the diff data.
> 
> I don't have numbers, so it is just hand-waving for my part.
> 
> > The ram bulk stage is where compression can do optimization,
> 
> If we do compression in the bulk page, xbzrle cache is going to be empty
> after that, so we need to re-send the whole page anyways (at least the 1st
> time).

Yes, your description is more exact. 

> > and beside
> > the ram bulk stage, xbzrle may do better than compression  in some
> > situation.
> 
> With your patch, there is no way to select xbzrle for bulk stage, and
> compression for the rest.  So this is is not an argument for this patch O:-)
> 
> > So
> > compression and xbzrle are not in conflict but complementary.
> 
> Oh, there is no conflict between them, I fully agree there.  The problem is
> when to use one or when to use the other.
> 
> My proposal:
> - user devices if it wants to use xbzrle or compression.  It is
>   completely clear what is going to happen.
> 
> - with this patch:
>   ram bulk stage use compressión and from there it uses xbzrle: this
>   needs at least to be documented on the man page and command line,
>   otherwise the user don't know.
> 
> - perhaps it is even a good idea to change the code to do
> 
>     if (is_zero_range(..))
>         send_it_as_one_byte()
>     if (it_is_on_bzrle_cache())
>         send as bzrle()
>     if (it is not on xbzrle_cache() {
>         put it on xbzrle_cache()
>         send it compressed()
>     }
> 
> And I am sure that there are even more posibilities.  The problem is that to
> choose one or another we should:
> - meassure what is better
> - decide what to implement
> - document how it works

I can image that XBZRLE will work better than compression in a situation where
only few bytes are changed in pages. :)

To be serious, I will do some test.

> As far as I can see, here we are doing the second item without having doing
> the 1st (or at least I haven't seen the results), and clearly without doing the
> third.
> 
> > I think it's a pity if we limit the use to select only one of them. If
> > there is no strong reason, I don't agree to drop this patch.
> 
> Then you need to at least add documentation to explain why/what you are
> doing.  If user selectcts xbzrle, it is clear what it does.  If user selects
> compress, it is also clear.  If it selects both, it is not clear (for looking at the
> documentation).

I am glad to add a document about that.

 
> Later, Juan.

Re: [Qemu-devel] [v5 08/12] migration: Add the core code of multi-thread compression

[Li, Liang Z]

> I thihnk this would make the code work, but not the locking.  You are using
> here:
> 
> quit_comp_thread:  global, and not completely clear what protects it
> comp_done_lock: global
> comp_done_cond: global
> 
> param[i].busy: I would suggest renaming to pending work
> param[i].mutex:
> param[i].cond:
>        thread is waiting for work
> 
> 
> Issues:
> 
> param->busy is protected on do_data_compress() and start_compression()
> with param->busy, but in flush_compressed_data() and
> comress_page_with_multithread() it is protected by comp_done_lock.
> 
> At this point, I would suggest to just drop param[i].mutex and use
> everywhere comp_done_lock.  We can make locking granularly later if
> needed, but 1st get it correct?
> Code basically does (forget termination and locking)
> 
> each compression_thread()
> 
>   while(1) {
>      while(!work_to_do)
>         wait_for_work
>      do_work
>   }
> 
> And the main thread does:
> 
> 
> while(1) {
>      foreacth compression_thread {
>           if thread free {
>              put it to work
>              break;
>           }
>           wait_for_thread_to_finish
>      }
> }
> 
> Notice how we are walking all threads each time that we need to do anything
> 
> Perhaps code should be more simple if we put the data that needs to be
> done on a global variable and change this to:
> 
> compression_thread
> 
>   while(1) {
>      while(!work_to_do)
>         wait_for_work
>      pick work from global variable
>      wakeup main thread
>      do_work
>   }
> 
> main thread:
> 
> put work on global variable
> while(nobody_pick_thework) {
>      signal all threads
>      wait for a compression thread to take the work }
> 
> Why?  because then we only have a global mutex and two condition variables,
> with a clear semantics:
> - lock protects two conditions and global variable with work
> - one condition where threads wait for work
> - one condition where main thread wait for a worker to be ready
> 
> As we would need to lock every single tame to put the work in the global
> variable, to wait or to pick up the work, we can stop all the:
> 
> if (!foo) {
>     mutex_lock
>     if(!foo) /* this time with lock */
>         ....
> }
> 
> 
> Sorry for the very long mail, if it makes you feel better, this is the second
> time that I wrote it, because the 1st version, my locking proposal didn't
> worked correctly.
> 
> What do you think?

I have tried to use comp_done_lock everywhere instead of param[i].mutex and found
that the performance is poor, the total migration time increase about 5 times.

So I add another variable to make the code correct and remain the param[i].mutex and the
logic of the compression thread and main thread.

Add another lock will drop the performance and increase total migration time ... 

Liang

> Later, Juan.

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Add the qmp and hmp commands to tune and query the parameters used in
> live migration.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

Hi,
   You might like to split this patch into two; one with no actual
parameters in the lists, and then a second one that adds your
compression options.

Dave

> ---
>  hmp-commands.hx               | 17 ++++++++
>  hmp.c                         | 56 +++++++++++++++++++++++++
>  hmp.h                         |  4 ++
>  include/migration/migration.h |  4 +-
>  migration/migration.c         | 96 +++++++++++++++++++++++++++++++++++++------
>  monitor.c                     | 25 +++++++++++
>  qapi-schema.json              | 86 ++++++++++++++++++++++++++++++++++++++
>  qmp-commands.hx               | 66 +++++++++++++++++++++++++++++
>  8 files changed, 339 insertions(+), 15 deletions(-)
> 
> diff --git a/hmp-commands.hx b/hmp-commands.hx
> index e37bc8b..ed0c06a 100644
> --- a/hmp-commands.hx
> +++ b/hmp-commands.hx
> @@ -985,6 +985,21 @@ Enable/Disable the usage of a capability @var{capability} for migration.
>  ETEXI
>  
>      {
> +        .name       = "migrate_set_parameter",
> +        .args_type  = "parameter:s,value:i",
> +        .params     = "parameter value",
> +        .help       = "Set the parameter for migration",
> +        .mhandler.cmd = hmp_migrate_set_parameter,
> +        .command_completion = migrate_set_parameter_completion,
> +    },
> +
> +STEXI
> +@item migrate_set_parameter @var{parameter} @var{value}
> +@findex migrate_set_parameter
> +Set the parameter @var{parameter} for migration.
> +ETEXI
> +
> +    {
>          .name       = "client_migrate_info",
>          .args_type  = "protocol:s,hostname:s,port:i?,tls-port:i?,cert-subject:s?",
>          .params     = "protocol hostname port tls-port cert-subject",
> @@ -1764,6 +1779,8 @@ show user network stack connection states
>  show migration status
>  @item info migrate_capabilities
>  show current migration capabilities
> +@item info migrate_parameters
> +show current migration parameters
>  @item info migrate_cache_size
>  show current migration XBZRLE cache size
>  @item info balloon
> diff --git a/hmp.c b/hmp.c
> index b47f331..1f67651 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -246,6 +246,27 @@ void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict)
>      qapi_free_MigrationCapabilityStatusList(caps);
>  }
>  
> +void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
> +{
> +    MigrationParameterStatusList *params, *p;
> +    MigrationParameterInt *data;
> +
> +    params = qmp_query_migrate_parameters(NULL);
> +
> +    if (params) {
> +        monitor_printf(mon, "parameters:");
> +        for (p = params; p; p = p->next) {
> +            data = (MigrationParameterInt *)p->value->data;
> +            monitor_printf(mon, " %s: %" PRId64,
> +                           MigrationParameter_lookup[p->value->kind],
> +                           data->value);
> +        }
> +        monitor_printf(mon, "\n");
> +    }
> +
> +    qapi_free_MigrationParameterStatusList(params);
> +}
> +
>  void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict)
>  {
>      monitor_printf(mon, "xbzrel cache size: %" PRId64 " kbytes\n",
> @@ -1140,6 +1161,41 @@ void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict)
>      }
>  }
>  
> +void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
> +{
> +    const char *param = qdict_get_str(qdict, "parameter");
> +    int value = qdict_get_int(qdict, "value");
> +    Error *err = NULL;
> +    MigrationParameterStatusList *params = g_malloc0(sizeof(*params));
> +    MigrationParameterInt *data;
> +    int i;
> +
> +    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +        if (strcmp(param, MigrationParameter_lookup[i]) == 0) {
> +            params->value = g_malloc0(sizeof(*params->value));
> +            params->value->kind = i;
> +            params->value->data = g_malloc0(sizeof(MigrationParameterInt));
> +            data = (MigrationParameterInt *)params->value->data;
> +            data->value = value;
> +            params->next = NULL;
> +            qmp_migrate_set_parameters(params, &err);
> +            break;
> +        }
> +    }
> +
> +    if (i == MIGRATION_PARAMETER_MAX) {
> +        error_set(&err, QERR_INVALID_PARAMETER, param);
> +    }
> +
> +    qapi_free_MigrationParameterStatusList(params);
> +
> +    if (err) {
> +        monitor_printf(mon, "migrate_set_parameter: %s\n",
> +                       error_get_pretty(err));
> +        error_free(err);
> +    }
> +}
> +
>  void hmp_set_password(Monitor *mon, const QDict *qdict)
>  {
>      const char *protocol  = qdict_get_str(qdict, "protocol");
> diff --git a/hmp.h b/hmp.h
> index 4bb5dca..b2b2d2c 100644
> --- a/hmp.h
> +++ b/hmp.h
> @@ -28,6 +28,7 @@ void hmp_info_chardev(Monitor *mon, const QDict *qdict);
>  void hmp_info_mice(Monitor *mon, const QDict *qdict);
>  void hmp_info_migrate(Monitor *mon, const QDict *qdict);
>  void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict);
> +void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict);
>  void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict);
>  void hmp_info_cpus(Monitor *mon, const QDict *qdict);
>  void hmp_info_block(Monitor *mon, const QDict *qdict);
> @@ -63,6 +64,7 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict);
> +void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_cache_size(Monitor *mon, const QDict *qdict);
>  void hmp_set_password(Monitor *mon, const QDict *qdict);
>  void hmp_expire_password(Monitor *mon, const QDict *qdict);
> @@ -111,6 +113,8 @@ void watchdog_action_completion(ReadLineState *rs, int nb_args,
>                                  const char *str);
>  void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
>                                         const char *str);
> +void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
> +                                      const char *str);
>  void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str);
>  void host_net_remove_completion(ReadLineState *rs, int nb_args,
>                                  const char *str);
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index 9ac1b23..434cc96 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -51,9 +51,7 @@ struct MigrationState
>      QEMUBH *cleanup_bh;
>      QEMUFile *file;
>      QemuThread *compress_thread;
> -    int compress_thread_count;
> -    int decompress_thread_count;
> -    int compress_level;
> +    int parameters[MIGRATION_PARAMETER_MAX];
>  
>      int state;
>      MigrationParams params;
> diff --git a/migration/migration.c b/migration/migration.c
> index 55f749e..89750ba 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -66,9 +66,12 @@ MigrationState *migrate_get_current(void)
>          .bandwidth_limit = MAX_THROTTLE,
>          .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
>          .mbps = -1,
> -        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> -        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
> -        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
> +        .parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] =
> +                DEFAULT_MIGRATE_COMPRESS_LEVEL,
> +        .parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
> +                DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> +        .parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
> +                DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
>      };
>  
>      return &current_migration;
> @@ -178,6 +181,33 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
>      return head;
>  }
>  
> +MigrationParameterStatusList *qmp_query_migrate_parameters(Error **errp)
> +{
> +    MigrationParameterStatusList *head = NULL;
> +    MigrationParameterStatusList *params;
> +    MigrationState *s = migrate_get_current();
> +    MigrationParameterInt *data;
> +    int i;
> +
> +    params = NULL; /* silence compiler warning */
> +    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +        if (head == NULL) {
> +            head = g_malloc0(sizeof(*params));
> +            params = head;
> +        } else {
> +            params->next = g_malloc0(sizeof(*params));
> +            params = params->next;
> +        }
> +        params->value = g_malloc(sizeof(*params->value));
> +        params->value->kind = i;
> +        params->value->data = g_malloc(sizeof(MigrationParameterInt));
> +        data = (MigrationParameterInt *)params->value->data;
> +        data->value = s->parameters[i];
> +    }
> +
> +    return head;
> +}
> +
>  static void get_xbzrle_cache_stats(MigrationInfo *info)
>  {
>      if (migrate_use_xbzrle()) {
> @@ -294,6 +324,44 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
>      }
>  }
>  
> +void qmp_migrate_set_parameters(MigrationParameterStatusList *params,
> +                                Error **errp)
> +{
> +    MigrationState *s = migrate_get_current();
> +    MigrationParameterStatusList *p;
> +    MigrationParameterInt *data;
> +
> +    for (p = params; p; p = p->next) {
> +        switch (p->value->kind) {
> +        case MIGRATION_PARAMETER_COMPRESS_LEVEL:
> +            data = (MigrationParameterInt *)p->value->data;
> +            if (data->value < 0 || data->value > 9) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
> +                          "is invalid, it should be in the range of 0 to 9");
> +                return;
> +            }
> +            break;
> +        case MIGRATION_PARAMETER_COMPRESS_THREADS:
> +        case MIGRATION_PARAMETER_DECOMPRESS_THREADS:
> +            if (s->state == MIG_STATE_ACTIVE || s->state == MIG_STATE_SETUP) {
> +                error_set(errp, QERR_MIGRATION_ACTIVE);
> +                return;
> +            }
> +            data = (MigrationParameterInt *)p->value->data;
> +            if (data->value < 1 || data->value > 255) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE,
> +                          "(de)compress_threads",
> +                          "is invalid, it should be in the range of 1 to 255");
> +                return;
> +            }
> +            break;
> +        default:
> +           return;
> +        }
> +        s->parameters[p->value->kind] = data->value;
> +    }
> +}
> +
>  /* shared migration helpers */
>  
>  static void migrate_set_state(MigrationState *s, int old_state, int new_state)
> @@ -400,9 +468,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
>      int64_t bandwidth_limit = s->bandwidth_limit;
>      bool enabled_capabilities[MIGRATION_CAPABILITY_MAX];
>      int64_t xbzrle_cache_size = s->xbzrle_cache_size;
> -    int compress_level = s->compress_level;
> -    int compress_thread_count = s->compress_thread_count;
> -    int decompress_thread_count = s->decompress_thread_count;
> +    int compress_level = s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
> +    int compress_thread_count =
> +            s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
> +    int decompress_thread_count =
> +            s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
>  
>      memcpy(enabled_capabilities, s->enabled_capabilities,
>             sizeof(enabled_capabilities));
> @@ -413,9 +483,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
>             sizeof(enabled_capabilities));
>      s->xbzrle_cache_size = xbzrle_cache_size;
>  
> -    s->compress_level = compress_level;
> -    s->compress_thread_count = compress_thread_count;
> -    s->decompress_thread_count = decompress_thread_count;
> +    s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
> +    s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
> +               compress_thread_count;
> +    s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
> +               decompress_thread_count;
>      s->bandwidth_limit = bandwidth_limit;
>      s->state = MIG_STATE_SETUP;
>      trace_migrate_set_state(MIG_STATE_SETUP);
> @@ -603,7 +675,7 @@ int migrate_compress_level(void)
>  
>      s = migrate_get_current();
>  
> -    return s->compress_level;
> +    return s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
>  }
>  
>  int migrate_compress_threads(void)
> @@ -612,7 +684,7 @@ int migrate_compress_threads(void)
>  
>      s = migrate_get_current();
>  
> -    return s->compress_thread_count;
> +    return s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
>  }
>  
>  int migrate_decompress_threads(void)
> @@ -621,7 +693,7 @@ int migrate_decompress_threads(void)
>  
>      s = migrate_get_current();
>  
> -    return s->decompress_thread_count;
> +    return s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
>  }
>  
>  int migrate_use_xbzrle(void)
> diff --git a/monitor.c b/monitor.c
> index c3cc060..499ae1c 100644
> --- a/monitor.c
> +++ b/monitor.c
> @@ -2873,6 +2873,13 @@ static mon_cmd_t info_cmds[] = {
>          .mhandler.cmd = hmp_info_migrate_capabilities,
>      },
>      {
> +        .name       = "migrate_parameters",
> +        .args_type  = "",
> +        .params     = "",
> +        .help       = "show current migration parameters",
> +        .mhandler.cmd = hmp_info_migrate_parameters,
> +    },
> +    {
>          .name       = "migrate_cache_size",
>          .args_type  = "",
>          .params     = "",
> @@ -4555,6 +4562,24 @@ void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
>      }
>  }
>  
> +void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
> +                                      const char *str)
> +{
> +    size_t len;
> +
> +    len = strlen(str);
> +    readline_set_completion_index(rs, len);
> +    if (nb_args == 2) {
> +        int i;
> +        for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +            const char *name = MigrationParameter_lookup[i];
> +            if (!strncmp(str, name, len)) {
> +                readline_add_completion(rs, name);
> +            }
> +        }
> +    }
> +}
> +
>  void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str)
>  {
>      int i;
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 0dfc4ce..5bf21fe 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -541,6 +541,92 @@
>  ##
>  { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
>  
> +# @MigrationParameter
> +#
> +# Migration parameters enumeration
> +#
> +# @compress-level: Set the compression level to be used in live migration,
> +#          the compression level is an integer between 0 and 9, where 0 means
> +#          no compression, 1 means the best compression speed, and 9 means best
> +#          compression ratio which will consume more CPU.
> +#
> +# @compress-threads: Set compression thread count to be used in live migration,
> +#          the compression thread count is an integer between 1 and 255.
> +#
> +# @decompress-threads: Set decompression thread count to be used in live
> +#          migration, the decompression thread count is an integer between 1
> +#          and 255. Usually, decompression is at least 4 times as fast as
> +#          compression, so set the decompress-threads to the number about 1/4
> +#          of compress-threads is adequate.
> +#
> +# Since: 2.3
> +##
> +{ 'enum': 'MigrationParameter',
> +  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
> +##
> +# @MigrationParameterBase
> +#
> +# Migration parameter information
> +#
> +# @parameter: the parameter of migration
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterBase',
> +  'data': {'parameter': 'MigrationParameter'} }
> +##
> +# @MigrationParameterInt
> +#
> +# Migration parameter information
> +#
> +# @value: parameter int
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterInt',
> +  'data': {'value': 'int'} }
> +##
> +# @MigrationParameterStatus
> +#
> +# Migration parameter information
> +#
> +# @compress-level: compression level
> +#
> +# @compress-threads: compression thread count
> +#
> +# @decompress-threads: decompression thread count
> +#
> +# Since: 2.3
> +##
> +{ 'union': 'MigrationParameterStatus',
> +  'base': 'MigrationParameterBase',
> +  'discriminator': 'parameter',
> +  'data': { 'compress-level': 'MigrationParameterInt',
> +            'compress-threads': 'MigrationParameterInt',
> +            'decompress-threads': 'MigrationParameterInt'} }
> +#
> +# @migrate-set-parameters
> +#
> +# Set the following migration parameters (like compress-level)
> +#
> +# @parameters: json array of parameter modifications to make
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'migrate-set-parameters',
> +  'data': { 'parameters': ['MigrationParameterStatus'] } }
> +##
> +# @query-migrate-parameters
> +#
> +# Returns information about the current migration parameters status
> +#
> +# Returns: @MigrationParametersStatus
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'query-migrate-parameters',
> +  'returns': ['MigrationParameterStatus'] }
> +##
>  ##
>  # @MouseInfo:
>  #
> diff --git a/qmp-commands.hx b/qmp-commands.hx
> index a85d847..2c4737b 100644
> --- a/qmp-commands.hx
> +++ b/qmp-commands.hx
> @@ -3295,6 +3295,72 @@ EQMP
>      },
>  
>  SQMP
> +migrate-set-parameters
> +----------------------
> +
> +Set migration parameters
> +
> +- "compress-level": set compression level during migration
> +- "compress-threads": set compression thread count for migration
> +- "decompress-threads": set decompression thread count for migration
> +
> +Arguments:
> +
> +Example:
> +
> +-> { "execute": "migrate-set-parameters" , "arguments":
> +     { "parameters": [ { "parameter": "compress-level", "value": 1 } ] } }
> +
> +EQMP
> +
> +    {
> +        .name       = "migrate-set-parameters",
> +        .args_type  = "parameters:O",
> +        .params     = "parameter:s,value:O",
> +	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
> +    },
> +SQMP
> +query-migrate-parameters
> +------------------------
> +
> +Query current migration parameters
> +
> +- "parameters": migration parameters value
> +         - "compress-level" : compression level value (json-int)
> +         - "compress-threads" : compression thread count value (json-int)
> +         - "decompress-threads" : decompression thread count value (json-int)
> +
> +Arguments:
> +
> +Example:
> +
> +-> { "execute": "query-migrate-parameters" }
> +<- {
> +      "return": [
> +         {
> +            "parameter": "compress-level",
> +            "value": 1
> +         },
> +         {
> +            "parameter": "compress-threads",
> +            "value": 8
> +         },
> +         {
> +            "parameter": "decompress-threads",
> +            "value": 2
> +         }
> +      ]
> +   }
> +
> +EQMP
> +
> +    {
> +        .name       = "query-migrate-parameters",
> +        .args_type  = "",
> +        .mhandler.cmd_new = qmp_marshal_input_query_migrate_parameters,
> +    },
> +
> +SQMP
>  query-balloon
>  -------------
>  
> -- 
> 1.9.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Markus Armbruster]

Copying Eric for additional QAPI design expertise.

I'm going to review just the QAPI schema changes.

Liang Li <liang.z.li@intel.com> writes:

> Add the qmp and hmp commands to tune and query the parameters used in
> live migration.
>
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
[...]
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 0dfc4ce..5bf21fe 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -541,6 +541,92 @@
>  ##
>  { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
>  
> +# @MigrationParameter
> +#
> +# Migration parameters enumeration
> +#
> +# @compress-level: Set the compression level to be used in live migration,
> +#          the compression level is an integer between 0 and 9, where 0 means
> +#          no compression, 1 means the best compression speed, and 9 means best
> +#          compression ratio which will consume more CPU.
> +#
> +# @compress-threads: Set compression thread count to be used in live migration,
> +#          the compression thread count is an integer between 1 and 255.
> +#
> +# @decompress-threads: Set decompression thread count to be used in live
> +#          migration, the decompression thread count is an integer between 1
> +#          and 255. Usually, decompression is at least 4 times as fast as
> +#          compression, so set the decompress-threads to the number about 1/4
> +#          of compress-threads is adequate.
> +#
> +# Since: 2.3
> +##
> +{ 'enum': 'MigrationParameter',
> +  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
> +##
> +# @MigrationParameterBase
> +#
> +# Migration parameter information
> +#
> +# @parameter: the parameter of migration
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterBase',
> +  'data': {'parameter': 'MigrationParameter'} }
> +##
> +# @MigrationParameterInt
> +#
> +# Migration parameter information
> +#
> +# @value: parameter int
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterInt',
> +  'data': {'value': 'int'} }
> +##
> +# @MigrationParameterStatus
> +#
> +# Migration parameter information
> +#
> +# @compress-level: compression level
> +#
> +# @compress-threads: compression thread count
> +#
> +# @decompress-threads: decompression thread count
> +#
> +# Since: 2.3
> +##
> +{ 'union': 'MigrationParameterStatus',
> +  'base': 'MigrationParameterBase',
> +  'discriminator': 'parameter',
> +  'data': { 'compress-level': 'MigrationParameterInt',
> +            'compress-threads': 'MigrationParameterInt',
> +            'decompress-threads': 'MigrationParameterInt'} }
> +#
> +# @migrate-set-parameters
> +#
> +# Set the following migration parameters (like compress-level)
> +#
> +# @parameters: json array of parameter modifications to make
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'migrate-set-parameters',
> +  'data': { 'parameters': ['MigrationParameterStatus'] } }

The command takes a list of key-value pairs.  Looks like this (example
stolen from your patch to qmp-commands.hx):

    { "execute": "migrate-set-parameters",
      "arguments": { "parameters":
                     [ { "parameter": "compress-level", "value": 1 } ] } }

Awkward.  I'd very much prefer

    { "execute": "migrate-set-parameters",
      "arguments": { "compress-level", 1 } }

I.e. the command simply takes the parameters as optional arguments.
Simpler, and a natural use of the schema language.

> +##
> +# @query-migrate-parameters
> +#
> +# Returns information about the current migration parameters status
> +#
> +# Returns: @MigrationParametersStatus
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'query-migrate-parameters',
> +  'returns': ['MigrationParameterStatus'] }
> +##
>  ##
>  # @MouseInfo:
>  #

Produces a list of key-value pairs.  Looks like this (stolen from the
same place):

    {
       "return": [
          {
             "parameter": "compress-level",
             "value": 1
          },
          {
             "parameter": "compress-threads",
             "value": 8
          },
          {
             "parameter": "decompress-threads",
             "value": 2
          }
       ]
    }

I'd very much prefer a simple object instead:

    { "return": { "compress-level": 1,
                  "compress-threads": 8,
                  "decompress-threads": 2 } }

[...]

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Li, Liang Z]

> > +#
> > +# Migration parameter information
> > +#
> > +# @compress-level: compression level
> > +#
> > +# @compress-threads: compression thread count # #
> > +@decompress-threads: decompression thread count # # Since: 2.3 ## {
> > +'union': 'MigrationParameterStatus',
> > +  'base': 'MigrationParameterBase',
> > +  'discriminator': 'parameter',
> > +  'data': { 'compress-level': 'MigrationParameterInt',
> > +            'compress-threads': 'MigrationParameterInt',
> > +            'decompress-threads': 'MigrationParameterInt'} } # #
> > +@migrate-set-parameters # # Set the following migration parameters
> > +(like compress-level) # # @parameters: json array of parameter
> > +modifications to make # # Since: 2.3 ## { 'command':
> > +'migrate-set-parameters',
> > +  'data': { 'parameters': ['MigrationParameterStatus'] } }
> 
> The command takes a list of key-value pairs.  Looks like this (example stolen
> from your patch to qmp-commands.hx):
> 
>     { "execute": "migrate-set-parameters",
>       "arguments": { "parameters":
>                      [ { "parameter": "compress-level", "value": 1 } ] } }
> 
> Awkward.  I'd very much prefer
> 
>     { "execute": "migrate-set-parameters",
>       "arguments": { "compress-level", 1 } }
> 
> I.e. the command simply takes the parameters as optional arguments.
> Simpler, and a natural use of the schema language.

Yes, it seems complicated.  Eric suggested to use this type of interface, because it can 
support different type of parameters if some new parameters will be added.  

> > +##
> > +# @query-migrate-parameters
> > +#
> > +# Returns information about the current migration parameters status #
> > +# Returns: @MigrationParametersStatus # # Since: 2.3 ## { 'command':
> > +'query-migrate-parameters',
> > +  'returns': ['MigrationParameterStatus'] } ##
> >  ##
> >  # @MouseInfo:
> >  #
> 
> Produces a list of key-value pairs.  Looks like this (stolen from the same
> place):
> 
>     {
>        "return": [
>           {
>              "parameter": "compress-level",
>              "value": 1
>           },
>           {
>              "parameter": "compress-threads",
>              "value": 8
>           },
>           {
>              "parameter": "decompress-threads",
>              "value": 2
>           }
>        ]
>     }
> 
> I'd very much prefer a simple object instead:
> 
>     { "return": { "compress-level": 1,
>                   "compress-threads": 8,
>                   "decompress-threads": 2 } }
> 

The same reason.

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Markus Armbruster]

"Li, Liang Z" <liang.z.li@intel.com> writes:

>> > +#
>> > +# Migration parameter information
>> > +#
>> > +# @compress-level: compression level
>> > +#
>> > +# @compress-threads: compression thread count # #
>> > +@decompress-threads: decompression thread count # # Since: 2.3 ## {
>> > +'union': 'MigrationParameterStatus',
>> > +  'base': 'MigrationParameterBase',
>> > +  'discriminator': 'parameter',
>> > +  'data': { 'compress-level': 'MigrationParameterInt',
>> > +            'compress-threads': 'MigrationParameterInt',
>> > +            'decompress-threads': 'MigrationParameterInt'} } # #
>> > +@migrate-set-parameters # # Set the following migration parameters
>> > +(like compress-level) # # @parameters: json array of parameter
>> > +modifications to make # # Since: 2.3 ## { 'command':
>> > +'migrate-set-parameters',
>> > +  'data': { 'parameters': ['MigrationParameterStatus'] } }
>> 
>> The command takes a list of key-value pairs.  Looks like this (example stolen
>> from your patch to qmp-commands.hx):
>> 
>>     { "execute": "migrate-set-parameters",
>>       "arguments": { "parameters":
>>                      [ { "parameter": "compress-level", "value": 1 } ] } }
>> 
>> Awkward.  I'd very much prefer
>> 
>>     { "execute": "migrate-set-parameters",
>>       "arguments": { "compress-level", 1 } }
>> 
>> I.e. the command simply takes the parameters as optional arguments.
>> Simpler, and a natural use of the schema language.
>
> Yes, it seems complicated.  Eric suggested to use this type of
> interface, because it can
> support different type of parameters if some new parameters will be added.  

I don't understand.

Schema for the three parameters we have now:

{ 'command': 'migrate-set-parameters',
  'data': { '*compression-level': 'int',
            '*compression-threads': 'int,
            '*decompression-threads': 'int' } }

Let's add a parameter with some other type:

{ 'command': 'migrate-set-parameters',
  'data': { '*compression-level': 'int',
            '*compression-threads': 'int,
            '*compression-algorithm': 'CompressionAlgorithm',
            '*decompression-threads': 'int', } }

CompressionAlgorithm could be an enum.

Why wouldn't that work?

>> > +##
>> > +# @query-migrate-parameters
>> > +#
>> > +# Returns information about the current migration parameters status #
>> > +# Returns: @MigrationParametersStatus # # Since: 2.3 ## { 'command':
>> > +'query-migrate-parameters',
>> > +  'returns': ['MigrationParameterStatus'] } ##
>> >  ##
>> >  # @MouseInfo:
>> >  #
>> 
>> Produces a list of key-value pairs.  Looks like this (stolen from the same
>> place):
>> 
>>     {
>>        "return": [
>>           {
>>              "parameter": "compress-level",
>>              "value": 1
>>           },
>>           {
>>              "parameter": "compress-threads",
>>              "value": 8
>>           },
>>           {
>>              "parameter": "decompress-threads",
>>              "value": 2
>>           }
>>        ]
>>     }
>> 
>> I'd very much prefer a simple object instead:
>> 
>>     { "return": { "compress-level": 1,
>>                   "compress-threads": 8,
>>                   "decompress-threads": 2 } }
>> 
>
> The same reason.

{ 'command': 'query-migrate-parameters',
  'returns': 'MigrationParameters' }

{ 'type': 'MigrationParameters',
  'data': { 'compression-level': 'int',
            'compression-threads': 'int,
            'decompression-threads': 'int' } }

Add one as above:

{ 'type': 'MigrationParameters',
  'data': { 'compression-level': 'int',
            'compression-threads': 'int,
            'compression-algorithm': 'CompressionAlgorithm',
            'decompression-threads': 'int', } }

Why wouldn't that work?

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Li, Liang Z]

> >> The command takes a list of key-value pairs.  Looks like this
> >> (example stolen from your patch to qmp-commands.hx):
> >>
> >>     { "execute": "migrate-set-parameters",
> >>       "arguments": { "parameters":
> >>                      [ { "parameter": "compress-level", "value": 1 }
> >> ] } }
> >>
> >> Awkward.  I'd very much prefer
> >>
> >>     { "execute": "migrate-set-parameters",
> >>       "arguments": { "compress-level", 1 } }
> >>
> >> I.e. the command simply takes the parameters as optional arguments.
> >> Simpler, and a natural use of the schema language.
> >
> > Yes, it seems complicated.  Eric suggested to use this type of
> > interface, because it can support different type of parameters if some
> > new parameters will be added.
> 
> I don't understand.
> 
> Schema for the three parameters we have now:
> 
> { 'command': 'migrate-set-parameters',
>   'data': { '*compression-level': 'int',
>             '*compression-threads': 'int,
>             '*decompression-threads': 'int' } }
> 
> Let's add a parameter with some other type:
> 
> { 'command': 'migrate-set-parameters',
>   'data': { '*compression-level': 'int',
>             '*compression-threads': 'int,
>             '*compression-algorithm': 'CompressionAlgorithm',
>             '*decompression-threads': 'int', } }
> 
> CompressionAlgorithm could be an enum.
> 
> Why wouldn't that work?

To be honest, I know very little about QMP and JSON. What should I do to make the 
schema as you described?

 
> >> > +##
> >> > +# @query-migrate-parameters
> >> > +#
> >> > +# Returns information about the current migration parameters
> >> > +status # # Returns: @MigrationParametersStatus # # Since: 2.3 ##
> { 'command':
> >> > +'query-migrate-parameters',
> >> > +  'returns': ['MigrationParameterStatus'] } ##
> >> >  ##
> >> >  # @MouseInfo:
> >> >  #
> >>
> >> Produces a list of key-value pairs.  Looks like this (stolen from the
> >> same
> >> place):
> >>
> >>     {
> >>        "return": [
> >>           {
> >>              "parameter": "compress-level",
> >>              "value": 1
> >>           },
> >>           {
> >>              "parameter": "compress-threads",
> >>              "value": 8
> >>           },
> >>           {
> >>              "parameter": "decompress-threads",
> >>              "value": 2
> >>           }
> >>        ]
> >>     }
> >>
> >> I'd very much prefer a simple object instead:
> >>
> >>     { "return": { "compress-level": 1,
> >>                   "compress-threads": 8,
> >>                   "decompress-threads": 2 } }
> >>
> >
> > The same reason.
> 
> { 'command': 'query-migrate-parameters',
>   'returns': 'MigrationParameters' }
> 
> { 'type': 'MigrationParameters',
>   'data': { 'compression-level': 'int',
>             'compression-threads': 'int,
>             'decompression-threads': 'int' } }
> 
> Add one as above:
> 
> { 'type': 'MigrationParameters',
>   'data': { 'compression-level': 'int',
>             'compression-threads': 'int,
>             'compression-algorithm': 'CompressionAlgorithm',
>             'decompression-threads': 'int', } }
> 
> Why wouldn't that work?

Could you be kind to point out what changes should I make?

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Eric Blake]

[-- Attachment #1: Type: text/plain, Size: 2009 bytes --]

On 03/18/2015 08:30 PM, Li, Liang Z wrote:

>>> +'migrate-set-parameters',
>>> +  'data': { 'parameters': ['MigrationParameterStatus'] } }
>>
>> The command takes a list of key-value pairs.  Looks like this (example stolen
>> from your patch to qmp-commands.hx):
>>
>>     { "execute": "migrate-set-parameters",
>>       "arguments": { "parameters":
>>                      [ { "parameter": "compress-level", "value": 1 } ] } }
>>
>> Awkward.  I'd very much prefer
>>
>>     { "execute": "migrate-set-parameters",
>>       "arguments": { "compress-level", 1 } }
>>
>> I.e. the command simply takes the parameters as optional arguments.
>> Simpler, and a natural use of the schema language.

Indeed, if we are going to add a new command, then we don't need to
worry about making it super-generic, and can avoid the nesting of
complex types.

> 
> Yes, it seems complicated.  Eric suggested to use this type of interface, because it can 
> support different type of parameters if some new parameters will be added.  

When I originally suggested complex nested types, I was suggesting that
we _reuse_ the existing migrate-set-capabilities (by making it set both
boolean and integer capabilities in one go).  If we do that, then we
need the struct complexity.  But since we are proposing a new command
rather than shoe-horning into an existing command, we might as well do
it cleanly to begin with.  So I like Markus' suggestion that we
eliminate some of the complexity and just go with specifying parameters
directly.


>> I'd very much prefer a simple object instead:
>>
>>     { "return": { "compress-level": 1,
>>                   "compress-threads": 8,
>>                   "decompress-threads": 2 } }
>>
> 
> The same reason.

I like the idea as well.  Sorry for misleading you into a more complex
initial implementation when a simpler will do.

-- 
Eric Blake   eblake redhat com    +1-919-301-3266
Libvirt virtualization library http://libvirt.org


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

Re: [Qemu-devel] [v5 12/12] migration: Add commands to set and query parameter

[Li, Liang Z]

> > >> The command takes a list of key-value pairs.  Looks like this
> > >> (example stolen from your patch to qmp-commands.hx):
> > >>
> > >>     { "execute": "migrate-set-parameters",
> > >>       "arguments": { "parameters":
> > >>                      [ { "parameter": "compress-level", "value": 1
> > >> } ] } }
> > >>
> > >> Awkward.  I'd very much prefer
> > >>
> > >>     { "execute": "migrate-set-parameters",
> > >>       "arguments": { "compress-level", 1 } }
> > >>
> > >> I.e. the command simply takes the parameters as optional arguments.
> > >> Simpler, and a natural use of the schema language.
> > >
> > > Yes, it seems complicated.  Eric suggested to use this type of
> > > interface, because it can support different type of parameters if
> > > some new parameters will be added.
> >
> > I don't understand.
> >
> > Schema for the three parameters we have now:
> >
> > { 'command': 'migrate-set-parameters',
> >   'data': { '*compression-level': 'int',
> >             '*compression-threads': 'int,
> >             '*decompression-threads': 'int' } }
> >
> > Let's add a parameter with some other type:
> >
> > { 'command': 'migrate-set-parameters',
> >   'data': { '*compression-level': 'int',
> >             '*compression-threads': 'int,
> >             '*compression-algorithm': 'CompressionAlgorithm',
> >             '*decompression-threads': 'int', } }
> >
> > CompressionAlgorithm could be an enum.
> >
> > Why wouldn't that work?
> 
> To be honest, I know very little about QMP and JSON. What should I do to
> make the schema as you described?
> 

You have answered me, sorry!