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

[Qemu-devel] [PATCH v7 0/14] 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 v6->v7
    
    -Change the core code of compression and decompression 
---
 arch_init.c                   | 502 ++++++++++++++++++++++++++++++++++++++++--
 hmp-commands.hx               |  17 ++
 hmp.c                         |  65 ++++++
 hmp.h                         |   4 +
 include/migration/migration.h |  10 +
 include/migration/qemu-file.h |   3 +
 migration/migration.c         | 122 ++++++++++
 migration/qemu-file.c         |  39 ++++
 monitor.c                     |  25 +++
 qapi-schema.json              |  79 ++++++-
 qmp-commands.hx               |  57 +++++
 11 files changed, 900 insertions(+), 23 deletions(-)
-- 
1.9.1

[Qemu-devel] [v7 01/14] 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>
Reviewed-by: Juan Quintela <quintela@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..3d477c3
--- /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 resource 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] [v7 02/14] 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>
Reviewed-by: Juan Quintela <quintela@redhat.com>
---
 arch_init.c                   | 90 ++++++++++++++++++++++++++++++++++++++++++-
 include/migration/migration.h |  8 ++++
 migration/migration.c         | 37 ++++++++++++++++++
 3 files changed, 133 insertions(+), 2 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 4c8fcee..c4e985e 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -316,6 +316,65 @@ 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 QemuThread *compress_threads;
+
+static void *do_data_compress(void *opaque)
+{
+    while (true) {
+
+    /* To be done */
+
+    }
+
+    return NULL;
+}
+
+static inline void terminate_compression_threads(void)
+{
+    /* To be done */
+}
+
+void migrate_compress_threads_join(void)
+{
+    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(compress_threads + i);
+    }
+    g_free(compress_threads);
+    g_free(comp_param);
+    compress_threads = NULL;
+    comp_param = NULL;
+}
+
+void migrate_compress_threads_create(void)
+{
+    int i, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    thread_count = migrate_compress_threads();
+    compress_threads = g_new0(QemuThread, thread_count);
+    comp_param = g_new0(CompressParam, thread_count);
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_create(compress_threads + i, "compress",
+                           do_data_compress, comp_param + i,
+                           QEMU_THREAD_JOINABLE);
+    }
+}
+
 /**
  * save_page_header: Write page header to wire
  *
@@ -693,6 +752,28 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
 }
 
 /**
+ * ram_save_compressed_page: compress the given page and send it to the stream
+ *
+ * Returns: Number of pages written.
+ *
+ * @f: QEMUFile where to send the data
+ * @block: block that contains the page we want to send
+ * @offset: offset inside the block for the page
+ * @last_stage: if we are at the completion stage
+ * @bytes_transferred: increase it with the number of transferred bytes
+ */
+static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
+                                    ram_addr_t offset, bool last_stage,
+                                    uint64_t *bytes_transferred)
+{
+    int pages = -1;
+
+    /* To be done*/
+
+    return pages;
+}
+
+/**
  * ram_find_and_save_block: Finds a dirty page and sends it to f
  *
  * Called within an RCU critical section.
@@ -733,8 +814,13 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage,
                 ram_bulk_stage = false;
             }
         } else {
-            pages = ram_save_page(f, block, offset, last_stage,
-                                  bytes_transferred);
+            if (migrate_use_compression()) {
+                pages = ram_save_compressed_page(f, block, offset, last_stage,
+                                                 bytes_transferred);
+            } else {
+                pages = ram_save_page(f, block, offset, last_stage,
+                                      bytes_transferred);
+            }
 
             /* if page is unmodified, continue to the next */
             if (pages > 0) {
diff --git a/include/migration/migration.h b/include/migration/migration.h
index bf09968..a3ebbf6 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -50,6 +50,8 @@ struct MigrationState
     QemuThread thread;
     QEMUBH *cleanup_bh;
     QEMUFile *file;
+    int compress_thread_count;
+    int compress_level;
 
     int state;
     MigrationParams params;
@@ -104,6 +106,8 @@ bool migration_has_finished(MigrationState *);
 bool migration_has_failed(MigrationState *);
 MigrationState *migrate_get_current(void);
 
+void migrate_compress_threads_create(void);
+void migrate_compress_threads_join(void);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -152,6 +156,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 bc42490..5a8b5a7 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -33,6 +33,11 @@
 #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)
 
@@ -52,6 +57,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;
@@ -305,6 +312,7 @@ static void migrate_fd_cleanup(void *opaque)
         qemu_thread_join(&s->thread);
         qemu_mutex_lock_iothread();
 
+        migrate_compress_threads_join();
         qemu_fclose(s->file);
         s->file = NULL;
     }
@@ -390,6 +398,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));
@@ -400,6 +410,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 = MIGRATION_STATUS_SETUP;
     trace_migrate_set_state(MIGRATION_STATUS_SETUP);
@@ -587,6 +599,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;
@@ -730,6 +766,7 @@ void migrate_fd_connect(MigrationState *s)
     /* Notify before starting migration thread */
     notifier_list_notify(&migration_state_notifiers, s);
 
+    migrate_compress_threads_create();
     qemu_thread_create(&s->thread, "migration", migration_thread, s,
                        QEMU_THREAD_JOINABLE);
 }
-- 
1.9.1

[Qemu-devel] [v7 03/14] 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>
Reviewed-by: Juan Quintela <quintela@redhat.com>
---
 arch_init.c                   | 74 +++++++++++++++++++++++++++++++++++++++++++
 include/migration/migration.h |  4 +++
 migration/migration.c         | 19 +++++++++++
 3 files changed, 97 insertions(+)

diff --git a/arch_init.c b/arch_init.c
index c4e985e..0c9f693 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>
@@ -127,6 +128,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;
@@ -321,8 +323,16 @@ struct CompressParam {
 };
 typedef struct CompressParam CompressParam;
 
+struct DecompressParam {
+    /* To be done */
+};
+typedef struct DecompressParam DecompressParam;
+
 static CompressParam *comp_param;
 static QemuThread *compress_threads;
+static DecompressParam *decomp_param;
+static QemuThread *decompress_threads;
+static uint8_t *compressed_data_buf;
 
 static void *do_data_compress(void *opaque)
 {
@@ -1199,10 +1209,57 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
     }
 }
 
+static void *do_data_decompress(void *opaque)
+{
+    while (true) {
+        /* To be done */
+    }
+
+    return NULL;
+}
+
+void migrate_decompress_threads_create(void)
+{
+    int i, thread_count;
+
+    thread_count = migrate_decompress_threads();
+    decompress_threads = g_new0(QemuThread, thread_count);
+    decomp_param = g_new0(DecompressParam, thread_count);
+    compressed_data_buf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
+    for (i = 0; i < thread_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;
+
+    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++;
 
@@ -1282,6 +1339,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 a3ebbf6..d4a1062 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -51,6 +51,7 @@ struct MigrationState
     QEMUBH *cleanup_bh;
     QEMUFile *file;
     int compress_thread_count;
+    int decompress_thread_count;
     int compress_level;
 
     int state;
@@ -108,6 +109,8 @@ MigrationState *migrate_get_current(void);
 
 void migrate_compress_threads_create(void);
 void migrate_compress_threads_join(void);
+void migrate_decompress_threads_create(void);
+void migrate_decompress_threads_join(void);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -159,6 +162,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 5a8b5a7..19409e6 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -35,6 +35,9 @@
 
 /* Default compression thread count */
 #define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+/* Default decompression thread count, usually decompression is at
+ * least 4 times as fast as compression.*/
+#define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
 /*0: means nocompress, 1: best speed, ... 9: best compress ratio */
 #define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
 
@@ -58,6 +61,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,
     };
 
@@ -113,6 +117,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();
@@ -121,6 +126,7 @@ static void process_incoming_migration_co(void *opaque)
     bdrv_invalidate_cache_all(&local_err);
     if (local_err) {
         error_report_err(local_err);
+        migrate_decompress_threads_join();
         exit(EXIT_FAILURE);
     }
 
@@ -129,6 +135,7 @@ static void process_incoming_migration_co(void *opaque)
     } else {
         runstate_set(RUN_STATE_PAUSED);
     }
+    migrate_decompress_threads_join();
 }
 
 void process_incoming_migration(QEMUFile *f)
@@ -137,6 +144,7 @@ void process_incoming_migration(QEMUFile *f)
     int fd = qemu_get_fd(f);
 
     assert(fd != -1);
+    migrate_decompress_threads_create();
     qemu_set_nonblock(fd);
     qemu_coroutine_enter(co, f);
 }
@@ -400,6 +408,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));
@@ -412,6 +421,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 = MIGRATION_STATUS_SETUP;
     trace_migrate_set_state(MIGRATION_STATUS_SETUP);
@@ -623,6 +633,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] [v7 04/14] 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>
Reviewed-by: Juan Quintela <quintela@redhat.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 745a850..a01c5b8 100644
--- a/include/migration/qemu-file.h
+++ b/include/migration/qemu-file.h
@@ -159,6 +159,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 1a4f986..2750365 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"
@@ -546,3 +547,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] [v7 05/14] 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>
Reviewed-by: Juan Quintela <quintela@redhat.com>
---
 arch_init.c | 37 ++++++++++++++++++++++++++++++++++++-
 1 file changed, 36 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 0c9f693..00d3a18 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -319,7 +319,14 @@ static uint32_t last_version;
 static bool ram_bulk_stage;
 
 struct CompressParam {
-    /* To be done */
+    bool start;
+    bool done;
+    bool quit;
+    QEMUFile *file;
+    QemuMutex mutex;
+    QemuCond cond;
+    RAMBlock *block;
+    ram_addr_t offset;
 };
 typedef struct CompressParam CompressParam;
 
@@ -330,6 +337,14 @@ typedef struct DecompressParam DecompressParam;
 
 static CompressParam *comp_param;
 static QemuThread *compress_threads;
+/* 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 DecompressParam *decomp_param;
 static QemuThread *decompress_threads;
 static uint8_t *compressed_data_buf;
@@ -361,11 +376,20 @@ void migrate_compress_threads_join(void)
     thread_count = migrate_compress_threads();
     for (i = 0; i < thread_count; i++) {
         qemu_thread_join(compress_threads + 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(compress_threads);
     g_free(comp_param);
+    g_free(comp_done_cond);
+    g_free(comp_done_lock);
     compress_threads = NULL;
     comp_param = NULL;
+    comp_done_cond = NULL;
+    comp_done_lock = NULL;
 }
 
 void migrate_compress_threads_create(void)
@@ -378,7 +402,18 @@ void migrate_compress_threads_create(void)
     thread_count = migrate_compress_threads();
     compress_threads = 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);
+        comp_param[i].done = true;
+        qemu_mutex_init(&comp_param[i].mutex);
+        qemu_cond_init(&comp_param[i].cond);
         qemu_thread_create(compress_threads + i, "compress",
                            do_data_compress, comp_param + i,
                            QEMU_THREAD_JOINABLE);
-- 
1.9.1

[Qemu-devel] [v7 06/14] 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>
Reviewed-by: Juan Quintela <quintela@redhat.com>
---
 arch_init.c | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 00d3a18..67f5a24 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -331,7 +331,13 @@ struct CompressParam {
 typedef struct CompressParam CompressParam;
 
 struct DecompressParam {
-    /* To be done */
+    bool start;
+    bool quit;
+    QemuMutex mutex;
+    QemuCond cond;
+    void *des;
+    uint8 *compbuf;
+    int len;
 };
 typedef struct DecompressParam DecompressParam;
 
@@ -1262,6 +1268,9 @@ void migrate_decompress_threads_create(void)
     decomp_param = g_new0(DecompressParam, thread_count);
     compressed_data_buf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
     for (i = 0; i < thread_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);
@@ -1275,6 +1284,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] [v7 07/14] migration: Split save_zero_page from ram_save_page

[Liang Li]

Split the function save_zero_page from ram_save_page so that we can
reuse it later.

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>
---
 arch_init.c | 61 +++++++++++++++++++++++++++++++++++++++++++------------------
 1 file changed, 43 insertions(+), 18 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 67f5a24..44ac002 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -714,6 +714,34 @@ static void migration_bitmap_sync(void)
 }
 
 /**
+ * save_zero_page: Send the zero page to the stream
+ *
+ * Returns: Number of pages written.
+ *
+ * @f: QEMUFile where to send the data
+ * @block: block that contains the page we want to send
+ * @offset: offset inside the block for the page
+ * @p: pointer to the page
+ * @bytes_transferred: increase it with the number of transferred bytes
+ */
+static int save_zero_page(QEMUFile *f, RAMBlock *block, ram_addr_t offset,
+                          uint8_t *p, uint64_t *bytes_transferred)
+{
+    int pages = -1;
+
+    if (is_zero_range(p, TARGET_PAGE_SIZE)) {
+        acct_info.dup_pages++;
+        *bytes_transferred += save_page_header(f, block,
+                                               offset | RAM_SAVE_FLAG_COMPRESS);
+        qemu_put_byte(f, 0);
+        *bytes_transferred += 1;
+        pages = 1;
+    }
+
+    return pages;
+}
+
+/**
  * ram_save_page: Send the given page to the stream
  *
  * Returns: Number of pages written.
@@ -761,25 +789,22 @@ 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_transferred += save_page_header(f, block,
-                                               offset | RAM_SAVE_FLAG_COMPRESS);
-        qemu_put_byte(f, 0);
-        *bytes_transferred += 1;
-        pages = 1;
-        /* 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()) {
-        pages = save_xbzrle_page(f, &p, current_addr, block,
-                                 offset, last_stage, bytes_transferred);
-        if (!last_stage) {
-            /* Can't send this cached data async, since the cache page
-             * might get updated before it gets to the wire
+    } else {
+        pages = save_zero_page(f, block, offset, p, bytes_transferred);
+        if (pages > 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()) {
+            pages = save_xbzrle_page(f, &p, current_addr, block,
+                                     offset, last_stage, bytes_transferred);
+            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] [v7 08/14] 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 | 183 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 177 insertions(+), 6 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 44ac002..d657a6c 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -355,12 +355,35 @@ static DecompressParam *decomp_param;
 static QemuThread *decompress_threads;
 static uint8_t *compressed_data_buf;
 
+static int do_compress_ram_page(CompressParam *param);
+
+static inline void notify_compression_done(CompressParam *param)
+{
+    qemu_mutex_lock(comp_done_lock);
+    param->done = true;
+    qemu_cond_signal(comp_done_cond);
+    qemu_mutex_unlock(comp_done_lock);
+}
+
 static void *do_data_compress(void *opaque)
 {
-    while (true) {
+    CompressParam *param = opaque;
 
-    /* To be done */
+    while (true) {
+        qemu_mutex_lock(&param->mutex);
+        while (!param->start && !param->quit) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+        }
+        if (param->quit) {
+            qemu_mutex_unlock(&param->mutex);
+            notify_compression_done(param);
+            break;
+        }
+        do_compress_ram_page(param);
+        param->start = false;
+        qemu_mutex_unlock(&param->mutex);
 
+        notify_compression_done(param);
     }
 
     return NULL;
@@ -368,7 +391,15 @@ static void *do_data_compress(void *opaque)
 
 static inline void terminate_compression_threads(void)
 {
-    /* To be done */
+    int idx, thread_count;
+
+    thread_count = migrate_compress_threads();
+    for (idx = 0; idx < thread_count; idx++) {
+        qemu_mutex_lock(&comp_param[idx].mutex);
+        comp_param[idx].quit = true;
+        qemu_cond_signal(&comp_param[idx].cond);
+        qemu_mutex_unlock(&comp_param[idx].mutex);
+    }
 }
 
 void migrate_compress_threads_join(void)
@@ -827,6 +858,95 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
     return pages;
 }
 
+static int do_compress_ram_page(CompressParam *param)
+{
+    int bytes_sent, blen;
+    uint8_t *p;
+    RAMBlock *block = param->block;
+    ram_addr_t offset = param->offset;
+
+    p = memory_region_get_ram_ptr(block->mr) + (offset & TARGET_PAGE_MASK);
+
+    bytes_sent = save_page_header(param->file, block, offset |
+                                  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)
+{
+    param->done = false;
+    qemu_mutex_lock(&param->mutex);
+    param->start = 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++) {
+        qemu_mutex_lock(comp_done_lock);
+        while (!comp_param[idx].done && !comp_param[idx].quit) {
+            qemu_cond_wait(comp_done_cond, comp_done_lock);
+        }
+        qemu_mutex_unlock(comp_done_lock);
+        if (!comp_param[idx].quit) {
+            len = qemu_put_qemu_file(f, comp_param[idx].file);
+            bytes_transferred += len;
+        }
+    }
+}
+
+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,
+                                           uint64_t *bytes_transferred)
+{
+    int idx, thread_count, bytes_xmit = -1, pages = -1;
+
+    thread_count = migrate_compress_threads();
+    qemu_mutex_lock(comp_done_lock);
+    while (true) {
+        for (idx = 0; idx < thread_count; idx++) {
+            if (comp_param[idx].done) {
+                bytes_xmit = qemu_put_qemu_file(f, comp_param[idx].file);
+                set_compress_params(&comp_param[idx], block, offset);
+                start_compression(&comp_param[idx]);
+                pages = 1;
+                *bytes_transferred += bytes_xmit;
+                break;
+            }
+        }
+        if (pages > 0) {
+            break;
+        } else {
+            qemu_cond_wait(comp_done_cond, comp_done_lock);
+        }
+    }
+    qemu_mutex_unlock(comp_done_lock);
+
+    return pages;
+}
+
 /**
  * ram_save_compressed_page: compress the given page and send it to the stream
  *
@@ -843,8 +963,59 @@ static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block,
                                     uint64_t *bytes_transferred)
 {
     int pages = -1;
+    uint64_t bytes_xmit;
+    MemoryRegion *mr = block->mr;
+    uint8_t *p;
+    int ret;
 
-    /* To be done*/
+    p = memory_region_get_ram_ptr(mr) + offset;
+
+    bytes_xmit = 0;
+    ret = ram_control_save_page(f, block->offset,
+                                offset, TARGET_PAGE_SIZE, &bytes_xmit);
+    if (bytes_xmit) {
+        *bytes_transferred += bytes_xmit;
+        pages = 1;
+    }
+    if (block == last_sent_block) {
+        offset |= RAM_SAVE_FLAG_CONTINUE;
+    }
+    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
+        if (ret != RAM_SAVE_CONTROL_DELAYED) {
+            if (bytes_xmit > 0) {
+                acct_info.norm_pages++;
+            } else if (bytes_xmit == 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);
+            pages = save_zero_page(f, block, offset, p, bytes_transferred);
+            if (pages == -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_xmit = do_compress_ram_page(&comp_param[0]);
+                qemu_put_qemu_file(f, comp_param[0].file);
+                *bytes_transferred += bytes_xmit;
+                pages = 1;
+            }
+        } else {
+            pages = save_zero_page(f, block, offset, p, bytes_transferred);
+            if (pages == -1) {
+                pages = compress_page_with_multi_thread(f, block, offset,
+                                                        bytes_transferred);
+            }
+        }
+    }
 
     return pages;
 }
@@ -912,8 +1083,6 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage,
     return pages;
 }
 
-static uint64_t bytes_transferred;
-
 void acct_update_position(QEMUFile *f, size_t size, bool zero)
 {
     uint64_t pages = size / TARGET_PAGE_SIZE;
@@ -1127,6 +1296,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
         }
         i++;
     }
+    flush_compressed_data(f);
     rcu_read_unlock();
 
     /*
@@ -1168,6 +1338,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
         }
     }
 
+    flush_compressed_data(f);
     ram_control_after_iterate(f, RAM_CONTROL_FINISH);
     migration_end();
 
-- 
1.9.1

[Qemu-devel] [v7 09/14] 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 | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index d657a6c..8fb2ea4 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -351,6 +351,8 @@ 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 compression_switch;
+
 static DecompressParam *decomp_param;
 static QemuThread *decompress_threads;
 static uint8_t *compressed_data_buf;
@@ -436,6 +438,7 @@ void migrate_compress_threads_create(void)
     if (!migrate_use_compression()) {
         return;
     }
+    compression_switch = true;
     thread_count = migrate_compress_threads();
     compress_threads = g_new0(QemuThread, thread_count);
     comp_param = g_new0(CompressParam, thread_count);
@@ -1059,9 +1062,16 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage,
                 block = QLIST_FIRST_RCU(&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()) {
                 pages = ram_save_compressed_page(f, block, offset, last_stage,
                                                  bytes_transferred);
             } else {
-- 
1.9.1

[Qemu-devel] [v7 10/14] 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 | 51 ++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 48 insertions(+), 3 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 8fb2ea4..b88e6b2 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -889,7 +889,6 @@ static inline void start_compression(CompressParam *param)
     qemu_mutex_unlock(&param->mutex);
 }
 
-
 static uint64_t bytes_transferred;
 
 static void flush_compressed_data(QEMUFile *f)
@@ -1458,8 +1457,28 @@ 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 (true) {
-        /* To be done */
+        qemu_mutex_lock(&param->mutex);
+        while (!param->start && !param->quit) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+            pagesize = TARGET_PAGE_SIZE;
+            /* 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->start = false;
+        }
+        if (param->quit) {
+            qemu_mutex_unlock(&param->mutex);
+            break;
+        }
+        qemu_mutex_unlock(&param->mutex);
     }
 
     return NULL;
@@ -1489,6 +1508,12 @@ void migrate_decompress_threads_join(void)
 
     thread_count = migrate_decompress_threads();
     for (i = 0; i < thread_count; i++) {
+        qemu_mutex_lock(&decomp_param[i].mutex);
+        decomp_param[i].quit = true;
+        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);
@@ -1505,7 +1530,27 @@ 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++) {
+            qemu_mutex_lock(&decomp_param[idx].mutex);
+            if (!decomp_param[idx].start) {
+                memcpy(decomp_param[idx].compbuf, compbuf, len);
+                decomp_param[idx].des = host;
+                decomp_param[idx].len = len;
+                decomp_param[idx].start = true;
+                qemu_cond_signal(&decomp_param[idx].cond);
+                qemu_mutex_unlock(&decomp_param[idx].mutex);
+                break;
+            }
+            qemu_mutex_unlock(&decomp_param[idx].mutex);
+        }
+        if (idx < thread_count) {
+            break;
+        }
+    }
 }
 
 static int ram_load(QEMUFile *f, void *opaque, int version_id)
-- 
1.9.1

[Qemu-devel] [v7 11/14] 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      | 11 ++++++++++-
 2 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index 19409e6..dc7db87 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -611,8 +611,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 ac9594d..b117008 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -515,13 +515,22 @@
 #          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. Please note that if compress and xbzrle are both
+#          on, compress only takes effect in the ram bulk stage, after that,
+#          it will be disabled and only xbzrle takes effect, this can help to
+#          minimize migration traffic. 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] [v7 12/14] migration: Use an array instead of 3 parameters

[Liang Li]

Put the three parameters related to multiple thread (de)compression
into an int array, and use an enum type to index the parameter.

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 include/migration/migration.h |  4 +---
 migration/migration.c         | 31 +++++++++++++++++++------------
 qapi-schema.json              | 23 +++++++++++++++++++++++
 3 files changed, 43 insertions(+), 15 deletions(-)

diff --git a/include/migration/migration.h b/include/migration/migration.h
index d4a1062..a6e025a 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -50,9 +50,7 @@ struct MigrationState
     QemuThread thread;
     QEMUBH *cleanup_bh;
     QEMUFile *file;
-    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 dc7db87..533717c 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -60,9 +60,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;
@@ -406,9 +409,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));
@@ -419,9 +424,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 = MIGRATION_STATUS_SETUP;
     trace_migrate_set_state(MIGRATION_STATUS_SETUP);
@@ -624,7 +631,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)
@@ -633,7 +640,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)
@@ -642,7 +649,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/qapi-schema.json b/qapi-schema.json
index b117008..e7e2343 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -569,6 +569,29 @@
 ##
 { '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'] }
+
 ##
 # @MouseInfo:
 #
-- 
1.9.1

[Qemu-devel] [v7 13/14] migration: Add qmp commands to set and query parameters

[Liang Li]

Add the qmp 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>
---
 migration/migration.c | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++
 qapi-schema.json      | 45 ++++++++++++++++++++++++++++++++++++++++
 qmp-commands.hx       | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 158 insertions(+)

diff --git a/migration/migration.c b/migration/migration.c
index 533717c..732d229 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -188,6 +188,21 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
     return head;
 }
 
+MigrationParameters *qmp_query_migrate_parameters(Error **errp)
+{
+    MigrationParameters *params;
+    MigrationState *s = migrate_get_current();
+
+    params = g_malloc0(sizeof(*params));
+    params->compress_level = s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
+    params->compress_threads =
+            s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
+    params->decompress_threads =
+            s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
+
+    return params;
+}
+
 static void get_xbzrle_cache_stats(MigrationInfo *info)
 {
     if (migrate_use_xbzrle()) {
@@ -301,6 +316,47 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
     }
 }
 
+void qmp_migrate_set_parameters(bool has_compress_level,
+                                int64_t compress_level,
+                                bool has_compress_threads,
+                                int64_t compress_threads,
+                                bool has_decompress_threads,
+                                int64_t decompress_threads, Error **errp)
+{
+    MigrationState *s = migrate_get_current();
+
+    if (has_compress_level && (compress_level < 0 || compress_level > 9)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
+                  "is invalid, it should be in the range of 0 to 9");
+        return;
+    }
+    if (has_compress_threads &&
+            (compress_threads < 1 || compress_threads > 255)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                  "compress_threads",
+                  "is invalid, it should be in the range of 1 to 255");
+        return;
+    }
+    if (has_decompress_threads &&
+            (decompress_threads < 1 || decompress_threads > 255)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                  "decompress_threads",
+                  "is invalid, it should be in the range of 1 to 255");
+        return;
+    }
+
+    if (has_compress_level) {
+        s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
+    }
+    if (has_compress_threads) {
+        s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] = compress_threads;
+    }
+    if (has_decompress_threads) {
+        s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+                                                    decompress_threads;
+    }
+}
+
 /* shared migration helpers */
 
 static void migrate_set_state(MigrationState *s, int old_state, int new_state)
diff --git a/qapi-schema.json b/qapi-schema.json
index e7e2343..92d1bba 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -592,6 +592,51 @@
 { 'enum': 'MigrationParameter',
   'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
 
+#
+# @migrate-set-parameters
+#
+# Set the following migration parameters
+#
+# @compress-level: compression level
+#
+# @compress-threads: compression thread count
+#
+# @decompress-threads: decompression thread count
+#
+# Since: 2.3
+##
+{ 'command': 'migrate-set-parameters',
+  'data': { '*compress-level': 'int',
+            '*compress-threads': 'int',
+            '*decompress-threads': 'int'} }
+
+#
+# @MigrationParameters
+#
+# @compress-level: compression level
+#
+# @compress-threads: compression thread count
+#
+# @decompress-threads: decompression thread count
+#
+# Since: 2.3
+##
+{ 'type': 'MigrationParameters',
+  'data': { 'compress-level': 'int',
+            'compress-threads': 'int',
+            'decompress-threads': 'int'} }
+##
+# @query-migrate-parameters
+#
+# Returns information about the current migration parameters
+#
+# Returns: @MigrationParameters
+#
+# Since: 2.3
+##
+{ 'command': 'query-migrate-parameters',
+  'returns': 'MigrationParameters' }
+
 ##
 # @MouseInfo:
 #
diff --git a/qmp-commands.hx b/qmp-commands.hx
index 3a42ad0..8fcf5a8 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -3320,6 +3320,63 @@ EQMP
     },
 
 SQMP
+migrate-set-parameters
+----------------------
+
+Set migration parameters
+
+- "compress-level": set compression level during migration (json-int)
+- "compress-threads": set compression thread count for migration (json-int)
+- "decompress-threads": set decompression thread count for migration (json-int)
+
+Arguments:
+
+Example:
+
+-> { "execute": "migrate-set-parameters" , "arguments":
+      { "compress-level": 1 } }
+
+EQMP
+
+    {
+        .name       = "migrate-set-parameters",
+        .args_type  =
+            "compress-level:i?,compress-threads:i?,decompress-threads:i?",
+	.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": {
+         "decompress-threads", 2,
+         "compress-threads", 8,
+         "compress-level", 1
+      }
+   }
+
+EQMP
+
+    {
+        .name       = "query-migrate-parameters",
+        .args_type  = "",
+        .mhandler.cmd_new = qmp_marshal_input_query_migrate_parameters,
+    },
+
+SQMP
 query-balloon
 -------------
 
-- 
1.9.1

[Qemu-devel] [v7 14/14] migration: Add hmp interface to set and query parameters

[Liang Li]

Add the hmp interface 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           | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 hmp.h           |  4 ++++
 monitor.c       | 25 ++++++++++++++++++++++
 4 files changed, 111 insertions(+)

diff --git a/hmp-commands.hx b/hmp-commands.hx
index 3089533..61ab0ab 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -993,6 +993,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",
@@ -1762,6 +1777,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 f31ae27..dcb3940 100644
--- a/hmp.c
+++ b/hmp.c
@@ -252,6 +252,29 @@ void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict)
     qapi_free_MigrationCapabilityStatusList(caps);
 }
 
+void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
+{
+    MigrationParameters *params;
+
+    params = qmp_query_migrate_parameters(NULL);
+
+    if (params) {
+        monitor_printf(mon, "parameters:");
+        monitor_printf(mon, " %s: %" PRId64,
+            MigrationParameter_lookup[MIGRATION_PARAMETER_COMPRESS_LEVEL],
+            params->compress_level);
+        monitor_printf(mon, " %s: %" PRId64,
+            MigrationParameter_lookup[MIGRATION_PARAMETER_COMPRESS_THREADS],
+            params->compress_threads);
+        monitor_printf(mon, " %s: %" PRId64,
+            MigrationParameter_lookup[MIGRATION_PARAMETER_DECOMPRESS_THREADS],
+            params->decompress_threads);
+        monitor_printf(mon, "\n");
+    }
+
+    qapi_free_MigrationParameters(params);
+}
+
 void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict)
 {
     monitor_printf(mon, "xbzrel cache size: %" PRId64 " kbytes\n",
@@ -1184,6 +1207,48 @@ 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;
+    bool has_compress_level = false;
+    bool has_compress_threads = false;
+    bool has_decompress_threads = false;
+    int i;
+
+    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+        if (strcmp(param, MigrationParameter_lookup[i]) == 0) {
+            switch (i) {
+            case MIGRATION_PARAMETER_COMPRESS_LEVEL:
+                has_compress_level = true;
+                break;
+            case MIGRATION_PARAMETER_COMPRESS_THREADS:
+                has_compress_threads = true;
+                break;
+            case MIGRATION_PARAMETER_DECOMPRESS_THREADS:
+                has_decompress_threads = true;
+                break;
+            }
+            qmp_migrate_set_parameters(has_compress_level, value,
+                                       has_compress_threads, value,
+                                       has_decompress_threads, value,
+                                       &err);
+            break;
+        }
+    }
+
+    if (i == MIGRATION_PARAMETER_MAX) {
+        error_set(&err, QERR_INVALID_PARAMETER, param);
+    }
+
+    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 2b9308b..bb7c1e8 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);
@@ -64,6 +65,7 @@ void hmp_migrate_incoming(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);
@@ -114,6 +116,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/monitor.c b/monitor.c
index 68873ec..6ad79e9 100644
--- a/monitor.c
+++ b/monitor.c
@@ -2859,6 +2859,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     = "",
@@ -4548,6 +4555,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;
-- 
1.9.1

Re: [Qemu-devel] [v7 08/14] 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>

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

Re: [Qemu-devel] [v7 09/14] 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>

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


>  /* The empty QEMUFileOps will be used by file in CompressParam */
>  static const QEMUFileOps empty_ops = { };
> +static bool compression_switch;
> +
>  static DecompressParam *decomp_param;
>  static QemuThread *decompress_threads;
>  static uint8_t *compressed_data_buf;
> @@ -436,6 +438,7 @@ void migrate_compress_threads_create(void)
>      if (!migrate_use_compression()) {
>          return;
>      }
> +    compression_switch = true;
>      thread_count = migrate_compress_threads();
>      compress_threads = g_new0(QemuThread, thread_count);
>      comp_param = g_new0(CompressParam, thread_count);
> @@ -1059,9 +1062,16 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage,
>                  block = QLIST_FIRST_RCU(&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;
> +                }

I still think that it should be better:
a- don't mix them (or)
b- if we mix them, just use compression always that we sent whole pages.
   if xbzrle is not able to compress a page, use compression after
   putting the page on the cache.  i.e. try first to set through xbzrle,
   and if that don't work, use compression if possible (after zecond
   iteration, of course).)


But as you are the one doing the code....

Later, Juan.

PD. Yes, it can be changed later.

Re: [Qemu-devel] [v7 10/14] 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>
> ---
>  arch_init.c | 51 ++++++++++++++++++++++++++++++++++++++++++++++++---
>  1 file changed, 48 insertions(+), 3 deletions(-)
>
> diff --git a/arch_init.c b/arch_init.c
> index 8fb2ea4..b88e6b2 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -889,7 +889,6 @@ static inline void start_compression(CompressParam *param)
>      qemu_mutex_unlock(&param->mutex);
>  }
>  
> -
>  static uint64_t bytes_transferred;
>  
>  static void flush_compressed_data(QEMUFile *f)
> @@ -1458,8 +1457,28 @@ 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 (true) {
> -        /* To be done */
> +        qemu_mutex_lock(&param->mutex);
> +        while (!param->start && !param->quit) {
> +            qemu_cond_wait(&param->cond, &param->mutex);

We don't check here if we have received a quit


> +            pagesize = TARGET_PAGE_SIZE;
> +            /* 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->start = false;
> +        }
> +        if (param->quit) {
> +            qemu_mutex_unlock(&param->mutex);
> +            break;
> +        }
> +        qemu_mutex_unlock(&param->mutex);
>      }
>  
>      return NULL;
> @@ -1489,6 +1508,12 @@ void migrate_decompress_threads_join(void)
>  
>      thread_count = migrate_decompress_threads();
>      for (i = 0; i < thread_count; i++) {
> +        qemu_mutex_lock(&decomp_param[i].mutex);
> +        decomp_param[i].quit = true;
> +        qemu_cond_signal(&decomp_param[i].cond);

We set quit and send a signal to wakeup the thread, but it will try to
uncompress a page.  Shouldn't we change the position of the param->quit
check for exit?  I think it should be inside the loop.

> +        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);
> @@ -1505,7 +1530,27 @@ 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++) {
> +            qemu_mutex_lock(&decomp_param[idx].mutex);
> +            if (!decomp_param[idx].start) {
> +                memcpy(decomp_param[idx].compbuf, compbuf, len);
> +                decomp_param[idx].des = host;
> +                decomp_param[idx].len = len;
> +                decomp_param[idx].start = true;
> +                qemu_cond_signal(&decomp_param[idx].cond);
> +                qemu_mutex_unlock(&decomp_param[idx].mutex);
> +                break;
> +            }
> +            qemu_mutex_unlock(&decomp_param[idx].mutex);
> +        }
> +        if (idx < thread_count) {
> +            break;
> +        }
> +    }
>  }
>  
>  static int ram_load(QEMUFile *f, void *opaque, int version_id)

Re: [Qemu-devel] [v7 11/14] migration: Add interface to control compression

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> 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>

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

Re: [Qemu-devel] [v7 12/14] migration: Use an array instead of 3 parameters

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Put the three parameters related to multiple thread (de)compression
> into an int array, and use an enum type to index the parameter.
>
> 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>


BTW, for the libvirt people, what do you think about moving the other
parameters here, and make the other accessors deprecated?

I mean here:
- xbzrle_cache_size
-> get mbps
-> set/get migration_speed
-> get/set migration_downtime
-> autoconverge?

You get the idea.

Later, Juan.

Re: [Qemu-devel] [v7 13/14] migration: Add qmp commands to set and query parameters

[Juan Quintela]

Liang Li <liang.z.li@intel.com> wrote:
> Add the qmp 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>

Eric, can you review this and the following patch?  I think they are
correct, but I preffer someone more versed on QMP to review them.

Thanks, Juan.

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

[Li, Liang Z]

> > 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>
> 
> Reviewed-by: Juan Quintela <quintela@redhat.com>
> 
> 
> >  /* The empty QEMUFileOps will be used by file in CompressParam */
> > static const QEMUFileOps empty_ops = { };
> > +static bool compression_switch;
> > +
> >  static DecompressParam *decomp_param;  static QemuThread
> > *decompress_threads;  static uint8_t *compressed_data_buf; @@ -436,6
> > +438,7 @@ void migrate_compress_threads_create(void)
> >      if (!migrate_use_compression()) {
> >          return;
> >      }
> > +    compression_switch = true;
> >      thread_count = migrate_compress_threads();
> >      compress_threads = g_new0(QemuThread, thread_count);
> >      comp_param = g_new0(CompressParam, thread_count); @@ -1059,9
> > +1062,16 @@ static int ram_find_and_save_block(QEMUFile *f, bool
> last_stage,
> >                  block = QLIST_FIRST_RCU(&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;
> > +                }
> 
> I still think that it should be better:
> a- don't mix them (or)
> b- if we mix them, just use compression always that we sent whole pages.
>    if xbzrle is not able to compress a page, use compression after
>    putting the page on the cache.  i.e. try first to set through xbzrle,
>    and if that don't work, use compression if possible (after zecond
>    iteration, of course).)
> 
> 
> But as you are the one doing the code....
> 

I prefer the b solution that you suggest, maybe I can send a patch later. But now, I don't want to
make a big changes before the patch being merged.

Liang

> Later, Juan.
> 
> PD. Yes, it can be changed later.

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

[Li, Liang Z]

> > @@ -889,7 +889,6 @@ static inline void
> start_compression(CompressParam *param)
> >      qemu_mutex_unlock(&param->mutex);  }
> >
> > -
> >  static uint64_t bytes_transferred;
> >
> >  static void flush_compressed_data(QEMUFile *f) @@ -1458,8 +1457,28
> @@
> > 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 (true) {
> > -        /* To be done */
> > +        qemu_mutex_lock(&param->mutex);
> > +        while (!param->start && !param->quit) {
> > +            qemu_cond_wait(&param->cond, &param->mutex);
> 
> We don't check here if we have received a quit
> 
> 
> > +            pagesize = TARGET_PAGE_SIZE;
> > +            /* 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->start = false;
> > +        }
> > +        if (param->quit) {
> > +            qemu_mutex_unlock(&param->mutex);
> > +            break;
> > +        }
> > +        qemu_mutex_unlock(&param->mutex);
> >      }
> >
> >      return NULL;
> > @@ -1489,6 +1508,12 @@ void migrate_decompress_threads_join(void)
> >
> >      thread_count = migrate_decompress_threads();
> >      for (i = 0; i < thread_count; i++) {
> > +        qemu_mutex_lock(&decomp_param[i].mutex);
> > +        decomp_param[i].quit = true;
> > +        qemu_cond_signal(&decomp_param[i].cond);
> 
> We set quit and send a signal to wakeup the thread, but it will try to
> uncompress a page.  Shouldn't we change the position of the param->quit
> check for exit?  I think it should be inside the loop.

Yes, you are right. if (param->quit)  should be put in front of uncompress().
 I will change it in the final version. (I really hope the next version is the final version :) )

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

[Juan Quintela]

"Li, Liang Z" <liang.z.li@intel.com> wrote:
>> > @@ -889,7 +889,6 @@ static inline void
>> start_compression(CompressParam *param)
>> >      qemu_mutex_unlock(&param->mutex);  }
>> >
>> > -
>> >  static uint64_t bytes_transferred;
>> >
>> >  static void flush_compressed_data(QEMUFile *f) @@ -1458,8 +1457,28
>> @@
>> > 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 (true) {
>> > -        /* To be done */
>> > +        qemu_mutex_lock(&param->mutex);
>> > +        while (!param->start && !param->quit) {
>> > +            qemu_cond_wait(&param->cond, &param->mutex);
>> 
>> We don't check here if we have received a quit
>> 
>> 
>> > +            pagesize = TARGET_PAGE_SIZE;
>> > +            /* 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->start = false;
>> > +        }
>> > +        if (param->quit) {
>> > +            qemu_mutex_unlock(&param->mutex);
>> > +            break;
>> > +        }
>> > +        qemu_mutex_unlock(&param->mutex);
>> >      }
>> >
>> >      return NULL;
>> > @@ -1489,6 +1508,12 @@ void migrate_decompress_threads_join(void)
>> >
>> >      thread_count = migrate_decompress_threads();
>> >      for (i = 0; i < thread_count; i++) {
>> > +        qemu_mutex_lock(&decomp_param[i].mutex);
>> > +        decomp_param[i].quit = true;
>> > +        qemu_cond_signal(&decomp_param[i].cond);
>> 
>> We set quit and send a signal to wakeup the thread, but it will try to
>> uncompress a page.  Shouldn't we change the position of the param->quit
>> check for exit?  I think it should be inside the loop.
>
> Yes, you are right. if (param->quit)  should be put in front of uncompress().
>  I will change it in the final version. (I really hope the next
> version is the final version :) )

I am sorry this took so long :p

Later, Juan.

Re: [Qemu-devel] [v7 13/14] migration: Add qmp commands to set and query parameters

[Li, Liang Z]

 
> Eric, can you review this and the following patch?  I think they are correct, but
> I preffer someone more versed on QMP to review them.
> 
> Thanks, Juan.

Hi Juan & Eric,

    Since the latest QEMU is 2.3.0-rc2, is it possible to merge this serial of patches to QEMU 2.3?

If it can't not, then I should change some version information in the .json file.

Liang

Re: [Qemu-devel] [v7 13/14] migration: Add qmp commands to set and query parameters

[Juan Quintela]

"Li, Liang Z" <liang.z.li@intel.com> wrote:
>  
>> Eric, can you review this and the following patch?  I think they are correct, but
>> I preffer someone more versed on QMP to review them.
>> 
>> Thanks, Juan.
>
> Hi Juan & Eric,
>
>     Since the latest QEMU is 2.3.0-rc2, is it possible to merge this
> serial of patches to QEMU 2.3?
>
> If it can't not, then I should change some version information in the .json file.

It looks like a too big change for -rc2, but it would be integrated at
the buginning of 2.4.0.

Later, Juan.

>
> Liang

Re: [Qemu-devel] [v7 11/14] migration: Add interface to control compression

[Eric Blake]

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

On 04/08/2015 12:20 AM, Liang Li wrote:
> 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      | 11 ++++++++++-
>  2 files changed, 15 insertions(+), 3 deletions(-)
> 

> +# @compress: Use multiple compression threads to accelerate live migration.
> +#          This feature can help to reduce the migration traffic, by sending
> +#          compressed pages. Please note that if compress and xbzrle are both
> +#          on, compress only takes effect in the ram bulk stage, after that,
> +#          it will be disabled and only xbzrle takes effect, this can help to
> +#          minimize migration traffic. The feature is disabled by default.
> +#          (since 2.3)

Maintainer can tweak this to 2.4, since we've missed 2.3.  My R-b still
stands.

-- 
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] [v7 12/14] migration: Use an array instead of 3 parameters

[Eric Blake]

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

On 04/08/2015 05:37 AM, Juan Quintela wrote:
> Liang Li <liang.z.li@intel.com> wrote:
>> Put the three parameters related to multiple thread (de)compression
>> into an int array, and use an enum type to index the parameter.
>>
>> 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>
> 
> 
> BTW, for the libvirt people, what do you think about moving the other
> parameters here, and make the other accessors deprecated?
> 
> I mean here:
> - xbzrle_cache_size
> -> get mbps
> -> set/get migration_speed
> -> get/set migration_downtime
> -> autoconverge?
> 
> You get the idea.

Setting multiple bits through one command is nicer. Libvirt will still
have to keep the code for driving older qemu, but I'm not opposed to
consolidation.

-- 
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] [v7 12/14] migration: Use an array instead of 3 parameters

[Eric Blake]

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

On 04/08/2015 12:20 AM, Liang Li wrote:
> Put the three parameters related to multiple thread (de)compression
> into an int array, and use an enum type to index the parameter.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  include/migration/migration.h |  4 +---
>  migration/migration.c         | 31 +++++++++++++++++++------------
>  qapi-schema.json              | 23 +++++++++++++++++++++++
>  3 files changed, 43 insertions(+), 15 deletions(-)
> 

> +++ b/qapi-schema.json
> @@ -569,6 +569,29 @@
>  ##
>  { '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.

s/so set/so setting/

> +#
> +# Since: 2.3

2.4

Maintainer can fix if a respin is not needed.

-- 
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] [v7 13/14] migration: Add qmp commands to set and query parameters

[Eric Blake]

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

On 04/08/2015 12:20 AM, Liang Li wrote:
> Add the qmp 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>
> ---
>  migration/migration.c | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++
>  qapi-schema.json      | 45 ++++++++++++++++++++++++++++++++++++++++
>  qmp-commands.hx       | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++
>  3 files changed, 158 insertions(+)
> 

> +void qmp_migrate_set_parameters(bool has_compress_level,
> +                                int64_t compress_level,
> +                                bool has_compress_threads,
> +                                int64_t compress_threads,
> +                                bool has_decompress_threads,
> +                                int64_t decompress_threads, Error **errp)
> +{
> +    MigrationState *s = migrate_get_current();
> +
> +    if (has_compress_level && (compress_level < 0 || compress_level > 9)) {
> +        error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
> +                  "is invalid, it should be in the range of 0 to 9");
> +        return;
> +    }
> +    if (has_compress_threads &&
> +            (compress_threads < 1 || compress_threads > 255)) {

Indentation is unusual.

> +        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
> +                  "compress_threads",
> +                  "is invalid, it should be in the range of 1 to 255");
> +        return;
> +    }
> +    if (has_decompress_threads &&
> +            (decompress_threads < 1 || decompress_threads > 255)) {

and again


> +++ b/qapi-schema.json
> @@ -592,6 +592,51 @@
>  { 'enum': 'MigrationParameter',
>    'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
>  
> +#
> +# @migrate-set-parameters
> +#
> +# Set the following migration parameters
> +#
> +# @compress-level: compression level
> +#
> +# @compress-threads: compression thread count
> +#
> +# @decompress-threads: decompression thread count
> +#
> +# Since: 2.3

2.4

> +##
> +{ 'command': 'migrate-set-parameters',
> +  'data': { '*compress-level': 'int',
> +            '*compress-threads': 'int',
> +            '*decompress-threads': 'int'} }
> +
> +#
> +# @MigrationParameters
> +#
> +# @compress-level: compression level
> +#
> +# @compress-threads: compression thread count
> +#
> +# @decompress-threads: decompression thread count
> +#
> +# Since: 2.3

2.4

> +##
> +{ 'type': 'MigrationParameters',
> +  'data': { 'compress-level': 'int',
> +            'compress-threads': 'int',
> +            'decompress-threads': 'int'} }
> +##
> +# @query-migrate-parameters
> +#
> +# Returns information about the current migration parameters
> +#
> +# Returns: @MigrationParameters
> +#
> +# Since: 2.3

2.4


>  SQMP
> +migrate-set-parameters
> +----------------------
> +
> +Set migration parameters
> +
> +- "compress-level": set compression level during migration (json-int)
> +- "compress-threads": set compression thread count for migration (json-int)
> +- "decompress-threads": set decompression thread count for migration (json-int)

Might be worth mentioning that on input, all three are optional; and
omitting them leaves the value unchanged, so mentioning the default is
also nice

Minor enough that I'm okay if the maintainer fixes the release number
and adds:
Reviewed-by: Eric Blake <eblake@redhat.com>

-- 
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] [v7 13/14] migration: Add qmp commands to set and query parameters

[Eric Blake]

On 04/14/2015 06:09 AM, Eric Blake wrote:
> On 04/08/2015 12:20 AM, Liang Li wrote:
>> Add the qmp 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>
>> ---
>>  migration/migration.c | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++
>>  qapi-schema.json      | 45 ++++++++++++++++++++++++++++++++++++++++
>>  qmp-commands.hx       | 57 +++++++++++++++++++++++++++++++++++++++++++++++++++
>>  3 files changed, 158 insertions(+)

>> +    if (has_compress_level && (compress_level < 0 || compress_level > 9)) {
>> +        error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
>> +                  "is invalid, it should be in the range of 0 to 9");

All three error messages are bogus; this expands to:
"Parameter 'compress_level' expects is invalid, it should be in the
range of 0 to 9"


> Minor enough that I'm okay if the maintainer fixes the release number
> and adds:
> Reviewed-by: Eric Blake <eblake@redhat.com>

I'm no longer okay with this patch; you'll need a v8 that improves the
error wording (QERR_ macros are seldom the right thing to use, you may
be better off open-coding error_setg() and doing a sane message in place).

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

Re: [Qemu-devel] [v7 14/14] migration: Add hmp interface to set and query parameters

[Eric Blake]

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

On 04/08/2015 12:20 AM, Liang Li wrote:
> Add the hmp interface 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           | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
>  hmp.h           |  4 ++++
>  monitor.c       | 25 ++++++++++++++++++++++
>  4 files changed, 111 insertions(+)
> 

Reviewed-by: Eric Blake <eblake@redhat.com>

-- 
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] [v7 12/14] migration: Use an array instead of 3 parameters

[Juan Quintela]

Eric Blake <eblake@redhat.com> wrote:
> On 04/08/2015 12:20 AM, Liang Li wrote:
>> Put the three parameters related to multiple thread (de)compression
>> into an int array, and use an enum type to index the parameter.
>> 
>> Signed-off-by: Liang Li <liang.z.li@intel.com>
>> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
>> ---
>>  include/migration/migration.h |  4 +---
>>  migration/migration.c         | 31 +++++++++++++++++++------------
>>  qapi-schema.json              | 23 +++++++++++++++++++++++
>>  3 files changed, 43 insertions(+), 15 deletions(-)
>> 
>
>> +++ b/qapi-schema.json
>> @@ -569,6 +569,29 @@
>>  ##
>>  { '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.
>
> s/so set/so setting/
>
>> +#
>> +# Since: 2.3
>
> 2.4
>
> Maintainer can fix if a respin is not needed.

Liang, I will fix the s/2.3/2.4/ while merging, no need of respin for
this.

Later, Juan.

Re: [Qemu-devel] [v7 12/14] migration: Use an array instead of 3 parameters

[Li, Liang Z]

> Eric Blake <eblake@redhat.com> wrote:
> > On 04/08/2015 12:20 AM, Liang Li wrote:
> >> Put the three parameters related to multiple thread (de)compression
> >> into an int array, and use an enum type to index the parameter.
> >>
> >> Signed-off-by: Liang Li <liang.z.li@intel.com>
> >> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> >> ---
> >>  include/migration/migration.h |  4 +---
> >>  migration/migration.c         | 31 +++++++++++++++++++------------
> >>  qapi-schema.json              | 23 +++++++++++++++++++++++
> >>  3 files changed, 43 insertions(+), 15 deletions(-)
> >>
> >
> >> +++ b/qapi-schema.json
> >> @@ -569,6 +569,29 @@
> >>  ##
> >>  { '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.
> >
> > s/so set/so setting/
> >
> >> +#
> >> +# Since: 2.3
> >
> > 2.4
> >
> > Maintainer can fix if a respin is not needed.
> 
> Liang, I will fix the s/2.3/2.4/ while merging, no need of respin for this.
> 
> Later, Juan.

OK, the v8 is very soon.

Liang