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

[Qemu-devel] [PATCH v3 0/13] 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 v2->v3
    
    -Splited the patch to 13 parts instead of 2
    -Rewrote the core code to do compression and decompression 
    -Updated the document
    -Added a common command interface to set and query parameters
    -Added some comments
    
-- 
1.9.1

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

[Liang Li]

Give some details about the multiple compression threads 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>
---
 docs/multi-thread-compression.txt | 141 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 141 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..3bbc641
--- /dev/null
+++ b/docs/multi-thread-compression.txt
@@ -0,0 +1,141 @@
+Use multiple thread (de)compression in live migration
+======================================================
+Copyright (C) 2014 Intel Corporation
+Author: Li Liang <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 migration time can also be reduced about
+70% in a typical case.
+
+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 reduced, this factor can reduce
+the migration time. If the process of the compression is quick
+enough, then the total migration time can be reduced, and multiple
+thread compression can be used to accelerate the compression process.
+
+The decompression speed of zlib is at least 4 times as quick as
+compression, if the source and destination CPU have equal speed,
+keeping the compression thread count 4 times the decompression
+thread count can avoid CPU waste.
+
+Compression level can be used to control the compression speed and the
+compression ratio. High compression ratio will take more time, level 0
+stands for no compression, level 1 stands for the best compression
+speed, and level 9 stands for the best compression ratio. Users can
+select a level number between 0 and 9.
+
+
+When to use the multiple thread compression in live migration
+==============================================================
+Compression of data will consume extra CPU cycles; so in a system with
+high overhead of CPU, avoid using this feature. When the network
+bandwidth is very limited and the CPU resource is adequate, use of
+multiple thread compression will be very helpful. If both the CPU and
+the network bandwidth are adequate, use of multiple thread compression
+can still help to reduce the migration time.
+
+Performance
+===========
+Test environment:
+
+CPU: Intel(R) Xeon(R) CPU E5-2680 0 @ 2.70GHz
+Socket Count: 2
+Ram: 128G
+NIC: Intel I350 (10/100/1000Mbps)
+Host OS: CentOS 7 64-bit
+Guest OS: Ubuntu 12.10 64-bit
+Parameter: qemu-system-x86_64 -enable-kvm -m 1024
+ /share/ia32e_ubuntu12.10.img -monitor stdio
+
+There is no additional application is running on the guest when doing
+the test.
+
+
+Speed limit: 32MB/s
+---------------------------------------------------------------
+                    | original  | compress thread: 8
+                    |   way     | decompress thread: 2
+                    |           | compression level: 1
+---------------------------------------------------------------
+total time(msec):   |  26561    |  7920
+---------------------------------------------------------------
+transferred ram(kB):|  877054   | 260641
+---------------------------------------------------------------
+throughput(mbps):   |  270.53   | 269.68
+---------------------------------------------------------------
+total ram(kB):      |  1057604  | 1057604
+---------------------------------------------------------------
+
+
+Speed limit: No
+---------------------------------------------------------------
+                    | original  | compress thread: 15
+                    |   way     | decompress thread: 4
+                    |           | compression level: 1
+---------------------------------------------------------------
+total time(msec):   |  7611     |  2888
+---------------------------------------------------------------
+transferred ram(kB):|  876761   | 262301
+---------------------------------------------------------------
+throughput(mbps):   |  943.78   | 744.27
+---------------------------------------------------------------
+total ram(kB):      |  1057604  | 1057604
+---------------------------------------------------------------
+
+Usage
+=====
+1. Verify both the source and destination QEMU are able
+to support the multiple thread compression migration:
+    {qemu} info_migrate_capablilites
+    {qemu} ... compress: off ...
+
+2. Activate compression on the souce:
+    {qemu} migrate_set_capability compress on
+
+3. Set the compression thread count on source:
+    {qemu} migrate_set_paramter 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 is the default setting:
+    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
+setting is not appropriate.
+
+TODO
+====
+Some faster (de)compression method such as lz4 and quicklz can help
+to reduce the CPU consumption when doing (de)compression. Less
+(de)compression threads are needed when doing the migration.
-- 
1.8.3.1

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

[Liang Li]

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 arch_init.c                   | 78 ++++++++++++++++++++++++++++++++++++++++++-
 include/migration/migration.h |  9 +++++
 migration.c                   | 38 +++++++++++++++++++++
 3 files changed, 124 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 7680d28..a988ec2 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -332,6 +332,67 @@ static uint64_t migration_dirty_pages;
 static uint32_t last_version;
 static bool ram_bulk_stage;
 
+struct compress_param {
+    /* To be done */
+};
+typedef struct compress_param compress_param;
+
+static compress_param *comp_param;
+static bool quit_thread;
+
+static void *do_data_compress(void *opaque)
+{
+    while (!quit_thread) {
+
+    /* To be done */
+
+    }
+    return NULL;
+}
+
+static inline void terminate_compression_threads(void)
+{
+    quit_thread = true;
+    /* To be done */
+}
+
+void migrate_compress_threads_join(MigrationState *s)
+{
+    int i, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    terminate_compression_threads();
+    thread_count = migrate_compress_threads();
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_join(s->compress_thread + i);
+    }
+    g_free(s->compress_thread);
+    g_free(comp_param);
+    s->compress_thread = NULL;
+    comp_param = NULL;
+}
+
+void migrate_compress_threads_create(MigrationState *s)
+{
+    int i, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    quit_thread = false;
+    thread_count = migrate_compress_threads();
+    s->compress_thread = g_malloc0(sizeof(QemuThread)
+        * thread_count);
+    comp_param = g_malloc0(sizeof(compress_param) * thread_count);
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_create(s->compress_thread + i, "compress",
+            do_data_compress, comp_param + i, QEMU_THREAD_JOINABLE);
+
+    }
+}
+
 /* Update the xbzrle cache to reflect a page that's been sent as all 0.
  * The important thing is that a stale (not-yet-0'd) page be replaced
  * by the new data.
@@ -643,6 +704,16 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
     return bytes_sent;
 }
 
+static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
+        ram_addr_t offset, bool last_stage)
+{
+    int bytes_sent = 0;
+
+    /* To be done*/
+
+    return bytes_sent;
+}
+
 /*
  * ram_find_and_save_block: Finds a page to send and sends it to f
  *
@@ -677,7 +748,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
                 ram_bulk_stage = false;
             }
         } else {
-            bytes_sent = ram_save_page(f, block, offset, last_stage);
+            if (migrate_use_compression()) {
+                bytes_sent = ram_save_compressed_page(f,
+                        block, offset, last_stage);
+            } else {
+                bytes_sent = ram_save_page(f, block, offset, last_stage);
+            }
 
             /* if page is unmodified, continue to the next */
             if (bytes_sent > 0) {
diff --git a/include/migration/migration.h b/include/migration/migration.h
index 3cb5ba8..daf6c81 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -49,6 +49,9 @@ struct MigrationState
     QemuThread thread;
     QEMUBH *cleanup_bh;
     QEMUFile *file;
+    QemuThread *compress_thread;
+    int compress_thread_count;
+    int compress_level;
 
     int state;
     MigrationParams params;
@@ -107,6 +110,8 @@ bool migration_has_finished(MigrationState *);
 bool migration_has_failed(MigrationState *);
 MigrationState *migrate_get_current(void);
 
+void migrate_compress_threads_create(MigrationState *s);
+void migrate_compress_threads_join(MigrationState *s);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -156,6 +161,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.c b/migration.c
index c49a05a..402daae 100644
--- a/migration.c
+++ b/migration.c
@@ -43,6 +43,11 @@ enum {
 #define BUFFER_DELAY     100
 #define XFER_LIMIT_RATIO (1000 / BUFFER_DELAY)
 
+/* Default compression thread count */
+#define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+/*0: means nocompress, 1: best speed, ... 9: best compress ratio */
+#define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
+
 /* Migration XBZRLE default cache size */
 #define DEFAULT_MIGRATE_CACHE_SIZE (64 * 1024 * 1024)
 
@@ -60,6 +65,8 @@ MigrationState *migrate_get_current(void)
         .bandwidth_limit = MAX_THROTTLE,
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
+        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
     };
 
     return &current_migration;
@@ -302,6 +309,7 @@ static void migrate_fd_cleanup(void *opaque)
         qemu_thread_join(&s->thread);
         qemu_mutex_lock_iothread();
 
+        migrate_compress_threads_join(s);
         qemu_fclose(s->file);
         s->file = NULL;
     }
@@ -373,6 +381,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));
@@ -383,6 +393,8 @@ static MigrationState *migrate_init(const MigrationParams *params)
            sizeof(enabled_capabilities));
     s->xbzrle_cache_size = xbzrle_cache_size;
 
+    s->compress_level = compress_level;
+    s->compress_thread_count = compress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -555,6 +567,31 @@ bool migrate_zero_blocks(void)
     return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS];
 }
 
+bool migrate_use_compression(void)
+{
+    /* Disable compression before the series of patches 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;
@@ -695,6 +732,7 @@ void migrate_fd_connect(MigrationState *s)
     /* Notify before starting migration thread */
     notifier_list_notify(&migration_state_notifiers, s);
 
+    migrate_compress_threads_create(s);
     qemu_thread_create(&s->thread, "migration", migration_thread, s,
                        QEMU_THREAD_JOINABLE);
 }
-- 
1.8.3.1

[Qemu-devel] [v3 03/13] migration: Add the framework of muti-thread decompression

[Liang Li]

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 arch_init.c                   | 70 +++++++++++++++++++++++++++++++++++++++++++
 include/migration/migration.h |  4 +++
 migration.c                   | 15 ++++++++++
 3 files changed, 89 insertions(+)

diff --git a/arch_init.c b/arch_init.c
index a988ec2..2f1d0c4 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -126,6 +126,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;
@@ -332,13 +333,26 @@ static uint64_t migration_dirty_pages;
 static uint32_t last_version;
 static bool ram_bulk_stage;
 
+/* using compressBound() to calculate the buffer size needed to save the
+ * compressed data, to support the maximun TARGET_PAGE_SIZE bytes of
+ * data, we need more 15 bytes, use 16 to align the data.
+ */
+#define COMPRESS_BUF_SIZE (TARGET_PAGE_SIZE + 16)
+
 struct compress_param {
     /* To be done */
 };
 typedef struct compress_param compress_param;
 
+struct decompress_param {
+    /* To be done */
+};
+typedef struct decompress_param decompress_param;
+
 static compress_param *comp_param;
 static bool quit_thread;
+static decompress_param *decomp_param;
+static QemuThread *decompress_threads;
 
 static void *do_data_compress(void *opaque)
 {
@@ -1124,10 +1138,54 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
     }
 }
 
+static void *do_data_decompress(void *opaque)
+{
+    while (!quit_thread) {
+        /* To be done */
+    }
+    return NULL;
+}
+
+void migrate_decompress_threads_create(int count)
+{
+    int i;
+
+    decompress_threads = g_malloc0(sizeof(QemuThread) * count);
+    decomp_param = g_malloc0(sizeof(decompress_param) * count);
+    quit_thread = false;
+    for (i = 0; i < count; i++) {
+        qemu_thread_create(decompress_threads + i, "decompress",
+            do_data_decompress, decomp_param + i, QEMU_THREAD_JOINABLE);
+    }
+}
+
+void migrate_decompress_threads_join(void)
+{
+    int i, thread_count;
+
+    quit_thread = true;
+    thread_count = migrate_decompress_threads();
+    for (i = 0; i < thread_count; i++) {
+        qemu_thread_join(decompress_threads + i);
+    }
+    g_free(decompress_threads);
+    g_free(decomp_param);
+    decompress_threads = NULL;
+    decomp_param = 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;
+    uint8_t compbuf[COMPRESS_BUF_SIZE];
 
     seq_iter++;
 
@@ -1201,6 +1259,18 @@ 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("Illegal RAM offset " RAM_ADDR_FMT, addr);
+                ret = -EINVAL;
+                break;
+            }
+
+            len = qemu_get_be32(f);
+            qemu_get_buffer(f, compbuf, len);
+            decompress_data_with_multi_threads(compbuf, 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 daf6c81..0c4f21c 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -51,6 +51,7 @@ struct MigrationState
     QEMUFile *file;
     QemuThread *compress_thread;
     int compress_thread_count;
+    int decompress_thread_count;
     int compress_level;
 
     int state;
@@ -112,6 +113,8 @@ MigrationState *migrate_get_current(void);
 
 void migrate_compress_threads_create(MigrationState *s);
 void migrate_compress_threads_join(MigrationState *s);
+void migrate_decompress_threads_create(int count);
+void migrate_decompress_threads_join(void);
 uint64_t ram_bytes_remaining(void);
 uint64_t ram_bytes_transferred(void);
 uint64_t ram_bytes_total(void);
@@ -164,6 +167,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.c b/migration.c
index 402daae..082ddb7 100644
--- a/migration.c
+++ b/migration.c
@@ -45,6 +45,7 @@ enum {
 
 /* Default compression thread count */
 #define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+#define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
 /*0: means nocompress, 1: best speed, ... 9: best compress ratio */
 #define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
 
@@ -66,6 +67,7 @@ MigrationState *migrate_get_current(void)
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
         .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
         .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
     };
 
@@ -123,10 +125,13 @@ static void process_incoming_migration_co(void *opaque)
     } else {
         runstate_set(RUN_STATE_PAUSED);
     }
+    migrate_decompress_threads_join();
 }
 
 void process_incoming_migration(QEMUFile *f)
 {
+    int thread_count = migrate_decompress_threads();
+    migrate_decompress_threads_create(thread_count);
     Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
     int fd = qemu_get_fd(f);
 
@@ -383,6 +388,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));
@@ -395,6 +401,7 @@ static MigrationState *migrate_init(const MigrationParams *params)
 
     s->compress_level = compress_level;
     s->compress_thread_count = compress_thread_count;
+    s->decompress_thread_count = decompress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -591,6 +598,14 @@ 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)
 {
-- 
1.8.3.1

[Qemu-devel] [v3 04/13] qemu-file: Add tow function will be used in migration

[Liang Li]

migrate_qemu_add_compression_data() compress the data
and put it to QEMUFile. migrate_qemu_flush() put the
data in the source QEMUFile to destination QEMUFile.

The two function can help to do live migration.

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

diff --git a/include/migration/qemu-file.h b/include/migration/qemu-file.h
index 401676b..d70fb8d 100644
--- a/include/migration/qemu-file.h
+++ b/include/migration/qemu-file.h
@@ -150,6 +150,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);
+size_t migrate_qemu_add_compression_data(QEMUFile *f,
+        const uint8_t *p, size_t size, int level);
+int migrate_qemu_flush(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/qemu-file.c b/qemu-file.c
index f938e36..2668ad0 100644
--- a/qemu-file.c
+++ b/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"
@@ -993,3 +994,34 @@ QEMUFile *qemu_bufopen(const char *mode, QEMUSizedBuffer *input)
     }
     return s->file;
 }
+
+/* compress size bytes of data start at p with specific compression
+ * leve and store the compressed data to the buffer of f.
+ */
+
+size_t migrate_qemu_add_compression_data(QEMUFile *f,
+        const uint8_t *p, size_t size, int level)
+{
+    size_t blen = IO_BUF_SIZE - f->buf_index - sizeof(int);
+
+    if (compress2(f->buf + f->buf_index + sizeof(int), &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(int);
+}
+
+int migrate_qemu_flush(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.8.3.1

[Qemu-devel] [v3 05/13] arch_init: alloc and free data struct in multi-thread compression

[Liang Li]

Define the data structure and varibles used when doing 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>
---
 arch_init.c | 34 +++++++++++++++++++++++++++++++++-
 1 file changed, 33 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index 2f1d0c4..f21a8ea 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -340,16 +340,29 @@ static bool ram_bulk_stage;
 #define COMPRESS_BUF_SIZE (TARGET_PAGE_SIZE + 16)
 
 struct compress_param {
-    /* To be done */
+    int state;
+    QEMUFile *file;
+    QemuMutex mutex;
+    QemuCond cond;
+    RAMBlock *block;
+    ram_addr_t offset;
 };
 typedef struct compress_param compress_param;
 
+enum {
+    DONE,
+    START,
+};
+
 struct decompress_param {
     /* To be done */
 };
 typedef struct decompress_param decompress_param;
 
 static compress_param *comp_param;
+static QemuMutex *mutex;
+static QemuCond *cond;
+static QEMUFileOps *empty_ops;
 static bool quit_thread;
 static decompress_param *decomp_param;
 static QemuThread *decompress_threads;
@@ -381,11 +394,22 @@ void migrate_compress_threads_join(MigrationState *s)
     thread_count = migrate_compress_threads();
     for (i = 0; i < thread_count; i++) {
         qemu_thread_join(s->compress_thread + i);
+        qemu_fclose(comp_param[i].file);
+        qemu_mutex_destroy(&comp_param[i].mutex);
+        qemu_cond_destroy(&comp_param[i].cond);
     }
+    qemu_mutex_destroy(mutex);
+    qemu_cond_destroy(cond);
     g_free(s->compress_thread);
     g_free(comp_param);
+    g_free(cond);
+    g_free(mutex);
+    g_free(empty_ops);
     s->compress_thread = NULL;
     comp_param = NULL;
+    cond = NULL;
+    mutex = NULL;
+    empty_ops = NULL;
 }
 
 void migrate_compress_threads_create(MigrationState *s)
@@ -400,7 +424,15 @@ void migrate_compress_threads_create(MigrationState *s)
     s->compress_thread = g_malloc0(sizeof(QemuThread)
         * thread_count);
     comp_param = g_malloc0(sizeof(compress_param) * thread_count);
+    cond = g_malloc0(sizeof(QemuCond));
+    mutex = g_malloc0(sizeof(QemuMutex));
+    empty_ops = g_malloc0(sizeof(QEMUFileOps));
+    qemu_cond_init(cond);
+    qemu_mutex_init(mutex);
     for (i = 0; i < thread_count; i++) {
+        comp_param[i].file = qemu_fopen_ops(NULL, empty_ops);
+        qemu_mutex_init(&comp_param[i].mutex);
+        qemu_cond_init(&comp_param[i].cond);
         qemu_thread_create(s->compress_thread + i, "compress",
             do_data_compress, comp_param + i, QEMU_THREAD_JOINABLE);
 
-- 
1.8.3.1

[Qemu-devel] [v3 06/13] arch_init: Add data struct used by decompression

[Liang Li]

Define the data structure and varibles used when doing 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>
---
 arch_init.c | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/arch_init.c b/arch_init.c
index f21a8ea..71cc756 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -355,7 +355,12 @@ enum {
 };
 
 struct decompress_param {
-    /* To be done */
+    int state;
+    QemuMutex mutex;
+    QemuCond cond;
+    void *des;
+    uint8 compbuf[COMPRESS_BUF_SIZE];
+    int len;
 };
 typedef struct decompress_param decompress_param;
 
@@ -1186,6 +1191,8 @@ void migrate_decompress_threads_create(int count)
     decomp_param = g_malloc0(sizeof(decompress_param) * count);
     quit_thread = false;
     for (i = 0; i < count; i++) {
+        qemu_mutex_init(&decomp_param[i].mutex);
+        qemu_cond_init(&decomp_param[i].cond);
         qemu_thread_create(decompress_threads + i, "decompress",
             do_data_decompress, decomp_param + i, QEMU_THREAD_JOINABLE);
     }
@@ -1199,6 +1206,8 @@ 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(decompress_threads);
     g_free(decomp_param);
-- 
1.8.3.1

[Qemu-devel] [v3 07/13] migraion: Rewrite the function ram_save_page()

[Liang Li]

We rewrite this function to reuse the code in it

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

diff --git a/arch_init.c b/arch_init.c
index 71cc756..0a575ed 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -596,6 +596,63 @@ static void migration_bitmap_sync_range(ram_addr_t start, ram_addr_t length)
     }
 }
 
+static int save_zero_and_xbzrle_page(QEMUFile *f, RAMBlock* block,
+        ram_addr_t offset, bool last_stage, bool *send_async)
+{
+    int bytes_sent;
+    int cont;
+    ram_addr_t current_addr;
+    MemoryRegion *mr = block->mr;
+    uint8_t *p;
+    int ret;
+
+    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
+
+    p = memory_region_get_ram_ptr(mr) + offset;
+
+    /* In doubt sent page as normal */
+    bytes_sent = -1;
+    ret = ram_control_save_page(f, block->offset,
+                           offset, TARGET_PAGE_SIZE, &bytes_sent);
+
+    XBZRLE_cache_lock();
+
+    current_addr = block->offset + offset;
+    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
+        if (ret != RAM_SAVE_CONTROL_DELAYED) {
+            if (bytes_sent > 0) {
+                acct_info.norm_pages++;
+            } else if (bytes_sent == 0) {
+                acct_info.dup_pages++;
+            }
+        }
+    } else if (is_zero_range(p, TARGET_PAGE_SIZE)) {
+        acct_info.dup_pages++;
+        bytes_sent = save_block_hdr(f, block, offset, cont,
+                                    RAM_SAVE_FLAG_COMPRESS);
+        qemu_put_byte(f, 0);
+        bytes_sent++;
+        /* Must let xbzrle know, otherwise a previous (now 0'd) cached
+         * page would be stale
+         */
+        xbzrle_cache_zero_page(current_addr);
+    } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
+        bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
+                                      offset, cont, last_stage);
+        if (!last_stage) {
+            /* Can't send this cached data async, since the cache page
+             * might get updated before it gets to the wire
+             */
+            if (send_async != NULL) {
+                *send_async = false;
+            }
+        }
+    }
+
+    XBZRLE_cache_unlock();
+
+    return bytes_sent;
+}
 
 /* Needs iothread lock! */
 /* Fix me: there are too many global variables used in migration process. */
@@ -691,55 +748,15 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
 {
     int bytes_sent;
     int cont;
-    ram_addr_t current_addr;
     MemoryRegion *mr = block->mr;
     uint8_t *p;
-    int ret;
     bool send_async = true;
 
-    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
-
-    p = memory_region_get_ram_ptr(mr) + offset;
-
-    /* In doubt sent page as normal */
-    bytes_sent = -1;
-    ret = ram_control_save_page(f, block->offset,
-                           offset, TARGET_PAGE_SIZE, &bytes_sent);
-
-    XBZRLE_cache_lock();
-
-    current_addr = block->offset + offset;
-    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
-        if (ret != RAM_SAVE_CONTROL_DELAYED) {
-            if (bytes_sent > 0) {
-                acct_info.norm_pages++;
-            } else if (bytes_sent == 0) {
-                acct_info.dup_pages++;
-            }
-        }
-    } else if (is_zero_range(p, TARGET_PAGE_SIZE)) {
-        acct_info.dup_pages++;
-        bytes_sent = save_block_hdr(f, block, offset, cont,
-                                    RAM_SAVE_FLAG_COMPRESS);
-        qemu_put_byte(f, 0);
-        bytes_sent++;
-        /* Must let xbzrle know, otherwise a previous (now 0'd) cached
-         * page would be stale
-         */
-        xbzrle_cache_zero_page(current_addr);
-    } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
-        bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
-                                      offset, cont, last_stage);
-        if (!last_stage) {
-            /* Can't send this cached data async, since the cache page
-             * might get updated before it gets to the wire
-             */
-            send_async = false;
-        }
-    }
-
-    /* XBZRLE overflow or normal page */
+    bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
+            last_stage, &send_async);
     if (bytes_sent == -1) {
+        cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
+        p = memory_region_get_ram_ptr(mr) + offset;
         bytes_sent = save_block_hdr(f, block, offset, cont, RAM_SAVE_FLAG_PAGE);
         if (send_async) {
             qemu_put_buffer_async(f, p, TARGET_PAGE_SIZE);
@@ -750,8 +767,6 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
         acct_info.norm_pages++;
     }
 
-    XBZRLE_cache_unlock();
-
     return bytes_sent;
 }
 
-- 
1.8.3.1

[Qemu-devel] [v3 08/13] migration: Add the core code of multi-thread compresion

[Liang Li]

At this point, multiple thread compression can't co-work with xbzrle.

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

diff --git a/arch_init.c b/arch_init.c
index 0a575ed..4109ad7 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -369,23 +369,43 @@ static QemuMutex *mutex;
 static QemuCond *cond;
 static QEMUFileOps *empty_ops;
 static bool quit_thread;
+static int one_byte_count;
 static decompress_param *decomp_param;
 static QemuThread *decompress_threads;
 
+static int do_compress_ram_page(compress_param *param);
+
 static void *do_data_compress(void *opaque)
 {
+    compress_param *param = opaque;
     while (!quit_thread) {
-
-    /* To be done */
-
+        qemu_mutex_lock(&param->mutex);
+        while (param->state != START) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+            if (quit_thread) {
+                break;
+            }
+            do_compress_ram_page(param);
+            qemu_mutex_lock(mutex);
+            param->state = DONE;
+            qemu_cond_signal(cond);
+            qemu_mutex_unlock(mutex);
+        }
+        qemu_mutex_unlock(&param->mutex);
     }
+
     return NULL;
 }
 
 static inline void terminate_compression_threads(void)
 {
+    int idx, thread_count;
+
+    thread_count = migrate_compress_threads();
     quit_thread = true;
-    /* To be done */
+    for (idx = 0; idx < thread_count; idx++) {
+        qemu_cond_signal(&comp_param[idx].cond);
+    }
 }
 
 void migrate_compress_threads_join(MigrationState *s)
@@ -770,13 +790,142 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
     return bytes_sent;
 }
 
+static int do_compress_ram_page(compress_param *param)
+{
+    int bytes_sent;
+    int blen = COMPRESS_BUF_SIZE;
+    int cont;
+    uint8_t *p;
+    RAMBlock *block = param->block;
+    ram_addr_t offset = param->offset;
+
+    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
+    p = memory_region_get_ram_ptr(block->mr) + offset;
+
+    bytes_sent = save_block_hdr(param->file, block,
+            offset, cont, RAM_SAVE_FLAG_COMPRESS_PAGE);
+    blen = migrate_qemu_add_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(compress_param *param)
+{
+    qemu_mutex_lock(&param->mutex);
+    param->state = START;
+    qemu_cond_signal(&param->cond);
+    qemu_mutex_unlock(&param->mutex);
+}
+
+
+static uint64_t bytes_transferred;
+
+static void flush_compressed_data(QEMUFile *f)
+{
+    int idx, len, thread_count;
+
+    if (!migrate_use_compression()) {
+        return;
+    }
+    thread_count = migrate_compress_threads();
+    for (idx = 0; idx < thread_count; idx++) {
+        if (comp_param[idx].state != DONE) {
+            qemu_mutex_lock(mutex);
+            while (comp_param[idx].state != DONE) {
+                qemu_cond_wait(cond, mutex);
+            }
+            qemu_mutex_unlock(mutex);
+        }
+        len = migrate_qemu_flush(f, comp_param[idx].file);
+        bytes_transferred += len;
+    }
+    if ((one_byte_count > 0) && (bytes_transferred > one_byte_count)) {
+        bytes_transferred -= one_byte_count;
+        one_byte_count = 0;
+    }
+}
+
+static inline void set_compress_params(compress_param *param,
+    RAMBlock *block, ram_addr_t offset)
+{
+    param->block = block;
+    param->offset = offset;
+}
+
+
+static int compress_page_with_multi_thread(QEMUFile *f,
+        RAMBlock *block, ram_addr_t offset)
+{
+    int idx, thread_count, bytes_sent = 0;
+
+    thread_count = migrate_compress_threads();
+    qemu_mutex_lock(mutex);
+    while (true) {
+        for (idx = 0; idx < thread_count; idx++) {
+            if (comp_param[idx].state == DONE) {
+                bytes_sent = migrate_qemu_flush(f, comp_param[idx].file);
+                set_compress_params(&comp_param[idx],
+                        block, offset);
+                start_compression(&comp_param[idx]);
+                if (bytes_sent == 0) {
+                    /* set bytes_sent to 1 in this case to prevent migration
+                     * from terminating, this 1 byte whill be added to
+                     * bytes_transferred later, minus 1 to keep the
+                     * bytes_transferred accurate */
+                    bytes_sent = 1;
+                    if (bytes_transferred <= 0) {
+                        one_byte_count++;
+                    } else {
+                        bytes_transferred -= 1;
+                    }
+                }
+                break;
+            }
+        }
+        if (bytes_sent > 0) {
+            break;
+        } else {
+            qemu_cond_wait(cond, mutex);
+        }
+    }
+    qemu_mutex_unlock(mutex);
+    return bytes_sent;
+}
+
 static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
         ram_addr_t offset, bool last_stage)
 {
     int bytes_sent = 0;
 
-    /* To be done*/
-
+    /* 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.
+     */
+    if (block != last_sent_block) {
+        flush_compressed_data(f);
+        bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
+                last_stage, NULL);
+        if (bytes_sent == -1) {
+            set_compress_params(&comp_param[0], block, offset);
+            /* Use the qemu thread to compress the data to make sure the
+             * first page is sent out before other pages
+             */
+            bytes_sent = do_compress_ram_page(&comp_param[0]);
+            if (bytes_sent > 0) {
+                migrate_qemu_flush(f, comp_param[0].file);
+            }
+        }
+    } else {
+        bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
+                last_stage, NULL);
+        if (bytes_sent == -1) {
+            bytes_sent = compress_page_with_multi_thread(f, block, offset);
+        }
+    }
     return bytes_sent;
 }
 
@@ -834,7 +983,6 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
     return bytes_sent;
 }
 
-static uint64_t bytes_transferred;
 
 void acct_update_position(QEMUFile *f, size_t size, bool zero)
 {
@@ -1043,6 +1191,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
         i++;
     }
 
+    flush_compressed_data(f);
     qemu_mutex_unlock_ramlist();
 
     /*
@@ -1089,6 +1238,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
         bytes_transferred += bytes_sent;
     }
 
+    flush_compressed_data(f);
     ram_control_after_iterate(f, RAM_CONTROL_FINISH);
     migration_end();
 
-- 
1.8.3.1

[Qemu-devel] [v3 09/13] 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>
---
 arch_init.c | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 4109ad7..14bc486 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -905,8 +905,11 @@ static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
      * block, and all the pages in last block should have been sent
      * out, keeping this order is important.
      */
-    if (block != last_sent_block) {
-        flush_compressed_data(f);
+    if ((!ram_bulk_stage && migrate_use_xbzrle()) ||
+                block != last_sent_block) {
+        if (block != last_sent_block) {
+            flush_compressed_data(f);
+        }
         bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
                 last_stage, NULL);
         if (bytes_sent == -1) {
@@ -961,6 +964,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
                 block = QTAILQ_FIRST(&ram_list.blocks);
                 complete_round = true;
                 ram_bulk_stage = false;
+                if (migrate_use_xbzrle()) {
+                    /* if xbzrle is on, we terminate the compression thread
+                     * at this point, there is no benefit from muti-thead */
+                    flush_compressed_data(f);
+                    terminate_compression_threads();
+                }
             }
         } else {
             if (migrate_use_compression()) {
-- 
1.8.3.1

[Qemu-devel] [v3 10/13] migration: Add the core code of multi-thread decompression

[Liang Li]

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

diff --git a/arch_init.c b/arch_init.c
index 14bc486..7103f4f 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>
@@ -820,6 +821,14 @@ static inline void start_compression(compress_param *param)
     qemu_mutex_unlock(&param->mutex);
 }
 
+static inline void start_decompression(decompress_param *param)
+{
+    qemu_mutex_lock(&param->mutex);
+    param->state = START;
+    qemu_cond_signal(&param->cond);
+    qemu_mutex_unlock(&param->mutex);
+}
+
 
 static uint64_t bytes_transferred;
 
@@ -1351,8 +1360,24 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
 
 static void *do_data_decompress(void *opaque)
 {
+    decompress_param *param = opaque;
     while (!quit_thread) {
-        /* To be done */
+        qemu_mutex_lock(&param->mutex);
+        while (param->state != START) {
+            qemu_cond_wait(&param->cond, &param->mutex);
+            if (quit_thread) {
+                break;
+            }
+            size_t pagesize = TARGET_PAGE_SIZE;
+            /* uncompress() will return failed in some case,
+             * especially when the page is dirted when doing
+             * the compression, ignore the return value because
+             * the dirty page will be retransferred. */
+            uncompress((Bytef *)param->des, &pagesize,
+                    (const Bytef *)param->compbuf, param->len);
+            param->state = DONE;
+        }
+        qemu_mutex_unlock(&param->mutex);
     }
     return NULL;
 }
@@ -1379,6 +1404,9 @@ void migrate_decompress_threads_join(void)
     quit_thread = true;
     thread_count = migrate_decompress_threads();
     for (i = 0; i < thread_count; i++) {
+        qemu_cond_signal(&decomp_param[i].cond);
+    }
+    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);
@@ -1392,7 +1420,23 @@ void migrate_decompress_threads_join(void)
 static void decompress_data_with_multi_threads(uint8_t *compbuf,
         void *host, int len)
 {
-    /* To be done */
+    int idx, thread_count;
+
+    thread_count = migrate_decompress_threads();
+    while (true) {
+        for (idx = 0; idx < thread_count; idx++) {
+            if (decomp_param[idx].state == DONE) {
+                memcpy(decomp_param[idx].compbuf, compbuf, len);
+                decomp_param[idx].des = host;
+                decomp_param[idx].len = len;
+                start_decompression(&decomp_param[idx]);
+                break;
+            }
+        }
+        if (idx < thread_count) {
+            break;
+        }
+    }
 }
 
 static int ram_load(QEMUFile *f, void *opaque, int version_id)
-- 
1.8.3.1

[Qemu-devel] [v3 11/13] migration: Add interface to control compression

[Liang Li]

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

Signed-off-by: Liang Li <liang.z.li@intel.com>
Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
---
 migration.c      | 7 +++++--
 qapi-schema.json | 6 +++++-
 2 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/migration.c b/migration.c
index 082ddb7..9d1613d 100644
--- a/migration.c
+++ b/migration.c
@@ -576,8 +576,11 @@ bool migrate_zero_blocks(void)
 
 bool migrate_use_compression(void)
 {
-    /* Disable compression before the series of patches 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 9ffdcf8..d371af3 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -491,13 +491,17 @@
 #          to enable the capability on the source VM. The feature is disabled by
 #          default. (since 1.6)
 #
+# @compress: Using the multiple compression threads to accelerate live migration.
+#          This feature can help to reduce the migration traffic, by sending
+#          compressed pages. The feature is disabled by default. (since 2.3)
+#
 # @auto-converge: If enabled, QEMU will automatically throttle down the guest
 #          to speed up convergence of RAM migration. (since 1.6)
 #
 # Since: 1.2
 ##
 { 'enum': 'MigrationCapability',
-  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks'] }
+  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks', 'compress'] }
 
 ##
 # @MigrationCapabilityStatus
-- 
1.8.3.1

[Qemu-devel] [v3 12/13] migration: Add command to set migration parameter

[Liang Li]

Add the qmp and hmp commands to tune 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               | 15 ++++++++++
 hmp.c                         | 32 +++++++++++++++++++++
 hmp.h                         |  3 ++
 include/migration/migration.h |  4 +--
 migration.c                   | 66 +++++++++++++++++++++++++++++++++++--------
 monitor.c                     | 18 ++++++++++++
 qapi-schema.json              | 44 +++++++++++++++++++++++++++++
 qmp-commands.hx               | 23 +++++++++++++++
 8 files changed, 190 insertions(+), 15 deletions(-)

diff --git a/hmp-commands.hx b/hmp-commands.hx
index e37bc8b..535b5ba 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -985,6 +985,21 @@ Enable/Disable the usage of a capability @var{capability} for migration.
 ETEXI
 
     {
+        .name       = "migrate_set_parameter",
+        .args_type  = "parameter:s,value:i",
+        .params     = "parameter value",
+        .help       = "Set the parameter for migration",
+        .mhandler.cmd = hmp_migrate_set_parameter,
+        .command_completion = migrate_set_parameter_completion,
+    },
+
+STEXI
+@item migrate_set_parameter @var{parameter} @var{value}
+@findex migrate_set_parameter
+Set the parameter @var{parameter} for migration.
+ETEXI
+
+    {
         .name       = "client_migrate_info",
         .args_type  = "protocol:s,hostname:s,port:i?,tls-port:i?,cert-subject:s?",
         .params     = "protocol hostname port tls-port cert-subject",
diff --git a/hmp.c b/hmp.c
index 63d7686..965c037 100644
--- a/hmp.c
+++ b/hmp.c
@@ -1079,6 +1079,38 @@ void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict)
     }
 }
 
+void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
+{
+    const char *param = qdict_get_str(qdict, "parameter");
+    int value = qdict_get_int(qdict, "value");
+    Error *err = NULL;
+    MigrationParameterStatusList *params = g_malloc0(sizeof(*params));
+    int i;
+
+    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+        if (strcmp(param, MigrationParameter_lookup[i]) == 0) {
+            params->value = g_malloc0(sizeof(*params->value));
+            params->value->parameter = i;
+            params->value->value = value;
+            params->next = NULL;
+            qmp_migrate_set_parameters(params, &err);
+            break;
+        }
+    }
+
+    if (i == MIGRATION_PARAMETER_MAX) {
+        error_set(&err, QERR_INVALID_PARAMETER, param);
+    }
+
+    qapi_free_MigrationParameterStatusList(params);
+
+    if (err) {
+        monitor_printf(mon, "migrate_set_parameter: %s\n",
+                       error_get_pretty(err));
+        error_free(err);
+    }
+}
+
 void hmp_set_password(Monitor *mon, const QDict *qdict)
 {
     const char *protocol  = qdict_get_str(qdict, "protocol");
diff --git a/hmp.h b/hmp.h
index 4bb5dca..bd1b203 100644
--- a/hmp.h
+++ b/hmp.h
@@ -63,6 +63,7 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict);
+void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict);
 void hmp_migrate_set_cache_size(Monitor *mon, const QDict *qdict);
 void hmp_set_password(Monitor *mon, const QDict *qdict);
 void hmp_expire_password(Monitor *mon, const QDict *qdict);
@@ -111,6 +112,8 @@ void watchdog_action_completion(ReadLineState *rs, int nb_args,
                                 const char *str);
 void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
                                        const char *str);
+void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
+                                       const char *str);
 void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str);
 void host_net_remove_completion(ReadLineState *rs, int nb_args,
                                 const char *str);
diff --git a/include/migration/migration.h b/include/migration/migration.h
index 0c4f21c..8e09b42 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -50,9 +50,7 @@ struct MigrationState
     QEMUBH *cleanup_bh;
     QEMUFile *file;
     QemuThread *compress_thread;
-    int compress_thread_count;
-    int decompress_thread_count;
-    int compress_level;
+    int parameters[MIGRATION_PARAMETER_MAX];
 
     int state;
     MigrationParams params;
diff --git a/migration.c b/migration.c
index 9d1613d..d3d377e 100644
--- a/migration.c
+++ b/migration.c
@@ -66,9 +66,12 @@ MigrationState *migrate_get_current(void)
         .bandwidth_limit = MAX_THROTTLE,
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
-        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
-        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
-        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
+        .parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] =
+                DEFAULT_MIGRATE_COMPRESS_LEVEL,
+        .parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
+                DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
+        .parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+                DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
     };
 
     return &current_migration;
@@ -292,6 +295,41 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
     }
 }
 
+void qmp_migrate_set_parameters(MigrationParameterStatusList *params,
+                                  Error **errp)
+{
+    MigrationState *s = migrate_get_current();
+    MigrationParameterStatusList *p;
+
+    for (p = params; p; p = p->next) {
+        switch (p->value->parameter) {
+        case MIGRATION_PARAMETER_COMPRESS_LEVEL:
+            if (p->value->value < 0 || p->value->value > 9) {
+                error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
+                    "is invalied, it should be in the rang of 0 to 9");
+                return;
+            }
+            break;
+        case MIGRATION_PARAMETER_COMPRESS_THREADS:
+        case MIGRATION_PARAMETER_DECOMPRESS_THREADS:
+            if (s->state == MIG_STATE_ACTIVE || s->state == MIG_STATE_SETUP) {
+                error_set(errp, QERR_MIGRATION_ACTIVE);
+                return;
+            }
+            if (p->value->value < 1 || p->value->value > 255) {
+                error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                    "(de)compress_threads",
+                    "is invalied, it should be in the rang of 1 to 255");
+                return;
+            }
+            break;
+        default:
+           return;
+        }
+        s->parameters[p->value->parameter] = p->value->value;
+    }
+}
+
 /* shared migration helpers */
 
 static void migrate_set_state(MigrationState *s, int old_state, int new_state)
@@ -386,9 +424,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));
@@ -399,9 +439,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
            sizeof(enabled_capabilities));
     s->xbzrle_cache_size = xbzrle_cache_size;
 
-    s->compress_level = compress_level;
-    s->compress_thread_count = compress_thread_count;
-    s->decompress_thread_count = decompress_thread_count;
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
+               compress_thread_count;
+    s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+               decompress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIG_STATE_SETUP;
     trace_migrate_set_state(MIG_STATE_SETUP);
@@ -589,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)
@@ -598,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)
@@ -607,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/monitor.c b/monitor.c
index f1031a1..4cf62b6 100644
--- a/monitor.c
+++ b/monitor.c
@@ -4544,6 +4544,24 @@ void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
     }
 }
 
+void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
+                                       const char *str)
+{
+    size_t len;
+
+    len = strlen(str);
+    readline_set_completion_index(rs, len);
+    if (nb_args == 2) {
+        int i;
+        for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+            const char *name = MigrationParameter_lookup[i];
+            if (!strncmp(str, name, len)) {
+                readline_add_completion(rs, name);
+            }
+        }
+    }
+}
+
 void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str)
 {
     int i;
diff --git a/qapi-schema.json b/qapi-schema.json
index d371af3..2caeccc 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -540,6 +540,50 @@
 ##
 { '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.
+#
+# Since: 2.3
+##
+{ 'enum': 'MigrationParameter',
+  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
+##
+# @MigrationParameterStatus
+#
+# Migration parameter information
+#
+# @parameter: parameter enum
+#
+# @value: parameter value int
+#
+# Since: 2.3
+##
+{ 'type': 'MigrationParameterStatus',
+  'data': { 'parameter' : 'MigrationParameter', 'value' : 'int' } }
+##
+# @migrate-set-parameters
+#
+# Set the following migration parameters (like compress-level)
+#
+# @parameters: json array of parameter modifications to make
+#
+# Since: 2.3
+##
+{ 'command': 'migrate-set-parameters',
+  'data': { 'parameters': ['MigrationParameterStatus'] } }
+##
 ##
 # @MouseInfo:
 #
diff --git a/qmp-commands.hx b/qmp-commands.hx
index 718dd92..59d2643 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -3225,6 +3225,29 @@ EQMP
     },
 
 SQMP
+migrate-set-parameters
+----------------------
+
+Set migration parameters
+
+- "compress-leve": multiple compression thread support
+
+Arguments:
+
+Example:
+
+-> { "execute": "migrate-set-parameters" , "arguments":
+     { "parameters": [ { "parameter": "compress-level", "value": 1 } ] } }
+
+EQMP
+
+    {
+        .name       = "migrate-set-parameters",
+        .args_type  = "parameters:O",
+        .params     = "parameter:s,value:i",
+	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
+    },
+SQMP
 query-balloon
 -------------
 
-- 
1.8.3.1

[Qemu-devel] [v3 13/13] migration: Add command to query migration parameter

[Liang Li]

Add the qmp and hmp commands to 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  |  2 ++
 hmp.c            | 19 +++++++++++++++++++
 hmp.h            |  1 +
 migration.c      | 25 +++++++++++++++++++++++++
 monitor.c        |  7 +++++++
 qapi-schema.json | 10 ++++++++++
 qmp-commands.hx  | 24 ++++++++++++++++++++++++
 7 files changed, 88 insertions(+)

diff --git a/hmp-commands.hx b/hmp-commands.hx
index 535b5ba..ed0c06a 100644
--- a/hmp-commands.hx
+++ b/hmp-commands.hx
@@ -1779,6 +1779,8 @@ show user network stack connection states
 show migration status
 @item info migrate_capabilities
 show current migration capabilities
+@item info migrate_parameters
+show current migration parameters
 @item info migrate_cache_size
 show current migration XBZRLE cache size
 @item info balloon
diff --git a/hmp.c b/hmp.c
index 965c037..b321b15 100644
--- a/hmp.c
+++ b/hmp.c
@@ -246,6 +246,25 @@ void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict)
     qapi_free_MigrationCapabilityStatusList(caps);
 }
 
+void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
+{
+    MigrationParameterStatusList *params, *p;
+
+    params = qmp_query_migrate_parameters(NULL);
+
+    if (params) {
+        monitor_printf(mon, "parameters: ");
+        for (p = params; p; p = p->next) {
+            monitor_printf(mon, "%s: %d ",
+                           MigrationParameter_lookup[p->value->parameter],
+                           (int)p->value->value);
+        }
+        monitor_printf(mon, "\n");
+    }
+
+    qapi_free_MigrationParameterStatusList(params);
+}
+
 void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict)
 {
     monitor_printf(mon, "xbzrel cache size: %" PRId64 " kbytes\n",
diff --git a/hmp.h b/hmp.h
index bd1b203..471417c 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);
diff --git a/migration.c b/migration.c
index d3d377e..f87aba3 100644
--- a/migration.c
+++ b/migration.c
@@ -179,6 +179,31 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
     return head;
 }
 
+MigrationParameterStatusList *qmp_query_migrate_parameters(Error **errp)
+{
+    MigrationParameterStatusList *head = NULL;
+    MigrationParameterStatusList *params;
+    MigrationState *s = migrate_get_current();
+    int i;
+
+    params = NULL; /* silence compiler warning */
+    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
+        if (head == NULL) {
+            head = g_malloc0(sizeof(*params));
+            params = head;
+        } else {
+            params->next = g_malloc0(sizeof(*params));
+            params = params->next;
+        }
+        params->value =
+            g_malloc(sizeof(*params->value));
+        params->value->parameter = i;
+        params->value->value = s->parameters[i];
+    }
+
+    return head;
+}
+
 static void get_xbzrle_cache_stats(MigrationInfo *info)
 {
     if (migrate_use_xbzrle()) {
diff --git a/monitor.c b/monitor.c
index 4cf62b6..6ab269f 100644
--- a/monitor.c
+++ b/monitor.c
@@ -2862,6 +2862,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     = "",
diff --git a/qapi-schema.json b/qapi-schema.json
index 2caeccc..ccdb6b2 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -584,6 +584,16 @@
 { 'command': 'migrate-set-parameters',
   'data': { 'parameters': ['MigrationParameterStatus'] } }
 ##
+# @query-migrate-parameters
+#
+# Returns information about the current migration parameters status
+#
+# Returns: @MigrationParametersStatus
+#
+# Since: 2.3
+##
+{ 'command': 'query-migrate-parameters', 'returns':   ['MigrationParameterStatus']}
+##
 ##
 # @MouseInfo:
 #
diff --git a/qmp-commands.hx b/qmp-commands.hx
index 59d2643..986eb95 100644
--- a/qmp-commands.hx
+++ b/qmp-commands.hx
@@ -3248,6 +3248,30 @@ EQMP
 	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
     },
 SQMP
+query-migrate-parameters
+------------------------
+
+Query current migration parameters
+
+- "parameters": migration parameters value
+         - "compress-level" : compress level value (json-int)
+
+Arguments:
+
+Example:
+
+-> { "execute": "query-migrate-parameters" }
+<- { "return": [ { "value": 1, "parameter": "compress-level" } ] }
+
+EQMP
+
+    {
+        .name       = "query-migrate-parameters",
+        .args_type  = "",
+        .mhandler.cmd_new = qmp_marshal_input_query_migrate_parameters,
+    },
+
+SQMP
 query-balloon
 -------------
 
-- 
1.8.3.1

Re: [Qemu-devel] [PATCH v3 0/13] migration: Add a new feature to do live migration

[Li, Liang Z]

> -----Original Message-----
> From: Li, Liang Z
> Sent: Friday, December 12, 2014 9:29 AM
> To: qemu-devel@nongnu.org
> Cc: quintela@redhat.com; lcapitulino@redhat.com; eblake@redhat.com;
> armbru@redhat.com; Zhang, Yang Z; dgilbert@redhat.com; Li, Liang Z
> Subject: [PATCH v3 0/13] migration: Add a new feature to do live migration
> 
> 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 v2->v3
> 
>     -Splited the patch to 13 parts instead of 2
>     -Rewrote the core code to do compression and decompression
>     -Updated the document
>     -Added a common command interface to set and query parameters
>     -Added some comments
> 
> --
> 1.9.1




Hi, all

	Could you help to review my patches?

Liang

Re: [Qemu-devel] [PATCH v3 0/13] migration: Add a new feature to do live migration

[Li, Liang Z]

Hi All, 

Have you  returned from  vacation?  If so, could you help to review my 
patches I have submitted three weeks ago? Thanks a lot.

Liang


> -----Original Message-----
> From: Li, Liang Z
> Sent: Friday, December 12, 2014 9:29 AM
> To: qemu-devel@nongnu.org
> Cc: quintela@redhat.com; lcapitulino@redhat.com; eblake@redhat.com;
> armbru@redhat.com; Zhang, Yang Z; dgilbert@redhat.com; Li, Liang Z
> Subject: [PATCH v3 0/13] migration: Add a new feature to do live migration
> 
> 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 v2->v3
> 
>     -Splited the patch to 13 parts instead of 2
>     -Rewrote the core code to do compression and decompression
>     -Updated the document
>     -Added a common command interface to set and query parameters
>     -Added some comments
> 
> --
> 1.9.1

Re: [Qemu-devel] [PATCH v3 0/13] migration: Add a new feature to do live migration

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> 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 v2->v3
>     
>     -Splited the patch to 13 parts instead of 2
>     -Rewrote the core code to do compression and decompression 
>     -Updated the document
>     -Added a common command interface to set and query parameters
>     -Added some comments

Hi,
  Apologies for taking so long; generally this is a lot nicer.
I'll reply to each patch individually, but two more general questions:
   a) Shouldn't compressBound be used somewhere to get the worst case compressed size?
   b) I think adding some trace points would be useful; it would make it easier to
      spot when you were waiting for threads etc.

Dave

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

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

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Give some details about the multiple compression threads and
> how to use it in live migration.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>

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

> ---
>  docs/multi-thread-compression.txt | 141 ++++++++++++++++++++++++++++++++++++++
>  1 file changed, 141 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..3bbc641
> --- /dev/null
> +++ b/docs/multi-thread-compression.txt
> @@ -0,0 +1,141 @@
> +Use multiple thread (de)compression in live migration
> +======================================================
> +Copyright (C) 2014 Intel Corporation
> +Author: Li Liang <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 migration time can also be reduced about
> +70% in a typical case.
> +
> +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 reduced, this factor can reduce
> +the migration time. If the process of the compression is quick
> +enough, then the total migration time can be reduced, and multiple
> +thread compression can be used to accelerate the compression process.
> +
> +The decompression speed of zlib is at least 4 times as quick as
> +compression, if the source and destination CPU have equal speed,
> +keeping the compression thread count 4 times the decompression
> +thread count can avoid CPU waste.
> +
> +Compression level can be used to control the compression speed and the
> +compression ratio. High compression ratio will take more time, level 0
> +stands for no compression, level 1 stands for the best compression
> +speed, and level 9 stands for the best compression ratio. Users can
> +select a level number between 0 and 9.
> +
> +
> +When to use the multiple thread compression in live migration
> +==============================================================
> +Compression of data will consume extra CPU cycles; so in a system with
> +high overhead of CPU, avoid using this feature. When the network
> +bandwidth is very limited and the CPU resource is adequate, use of
> +multiple thread compression will be very helpful. If both the CPU and
> +the network bandwidth are adequate, use of multiple thread compression
> +can still help to reduce the migration time.
> +
> +Performance
> +===========
> +Test environment:
> +
> +CPU: Intel(R) Xeon(R) CPU E5-2680 0 @ 2.70GHz
> +Socket Count: 2
> +Ram: 128G
> +NIC: Intel I350 (10/100/1000Mbps)
> +Host OS: CentOS 7 64-bit
> +Guest OS: Ubuntu 12.10 64-bit
> +Parameter: qemu-system-x86_64 -enable-kvm -m 1024
> + /share/ia32e_ubuntu12.10.img -monitor stdio
> +
> +There is no additional application is running on the guest when doing
> +the test.
> +
> +
> +Speed limit: 32MB/s
> +---------------------------------------------------------------
> +                    | original  | compress thread: 8
> +                    |   way     | decompress thread: 2
> +                    |           | compression level: 1
> +---------------------------------------------------------------
> +total time(msec):   |  26561    |  7920
> +---------------------------------------------------------------
> +transferred ram(kB):|  877054   | 260641
> +---------------------------------------------------------------
> +throughput(mbps):   |  270.53   | 269.68
> +---------------------------------------------------------------
> +total ram(kB):      |  1057604  | 1057604
> +---------------------------------------------------------------
> +
> +
> +Speed limit: No
> +---------------------------------------------------------------
> +                    | original  | compress thread: 15
> +                    |   way     | decompress thread: 4
> +                    |           | compression level: 1
> +---------------------------------------------------------------
> +total time(msec):   |  7611     |  2888
> +---------------------------------------------------------------
> +transferred ram(kB):|  876761   | 262301
> +---------------------------------------------------------------
> +throughput(mbps):   |  943.78   | 744.27
> +---------------------------------------------------------------
> +total ram(kB):      |  1057604  | 1057604
> +---------------------------------------------------------------
> +
> +Usage
> +=====
> +1. Verify both the source and destination QEMU are able
> +to support the multiple thread compression migration:
> +    {qemu} info_migrate_capablilites
> +    {qemu} ... compress: off ...
> +
> +2. Activate compression on the souce:
> +    {qemu} migrate_set_capability compress on
> +
> +3. Set the compression thread count on source:
> +    {qemu} migrate_set_paramter 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 is the default setting:
> +    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
> +setting is not appropriate.
> +
> +TODO
> +====
> +Some faster (de)compression method such as lz4 and quicklz can help
> +to reduce the CPU consumption when doing (de)compression. Less
> +(de)compression threads are needed when doing the migration.
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

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

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c                   | 78 ++++++++++++++++++++++++++++++++++++++++++-
>  include/migration/migration.h |  9 +++++
>  migration.c                   | 38 +++++++++++++++++++++
>  3 files changed, 124 insertions(+), 1 deletion(-)

Just a few minor comments fix below.

> diff --git a/arch_init.c b/arch_init.c
> index 7680d28..a988ec2 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -332,6 +332,67 @@ static uint64_t migration_dirty_pages;
>  static uint32_t last_version;
>  static bool ram_bulk_stage;
>  
> +struct compress_param {
> +    /* To be done */
> +};
> +typedef struct compress_param compress_param;

The struct types should be CompressParam rather than compress_param (like QemuThread - see the CODING_STYLE).

> +static compress_param *comp_param;
> +static bool quit_thread;
> +
> +static void *do_data_compress(void *opaque)
> +{
> +    while (!quit_thread) {
> +
> +    /* To be done */
> +
> +    }
> +    return NULL;
> +}

Initially I was worried if these needed atomics or barriers
for the 'quit_thread' - but later you add mutexs/cond that
I think should be enough.

> +
> +static inline void terminate_compression_threads(void)
> +{
> +    quit_thread = true;
> +    /* To be done */
> +}
> +
> +void migrate_compress_threads_join(MigrationState *s)
> +{
> +    int i, thread_count;
> +
> +    if (!migrate_use_compression()) {
> +        return;
> +    }
> +    terminate_compression_threads();
> +    thread_count = migrate_compress_threads();
> +    for (i = 0; i < thread_count; i++) {
> +        qemu_thread_join(s->compress_thread + i);
> +    }
> +    g_free(s->compress_thread);
> +    g_free(comp_param);
> +    s->compress_thread = NULL;
> +    comp_param = NULL;
> +}
> +
> +void migrate_compress_threads_create(MigrationState *s)
> +{
> +    int i, thread_count;
> +
> +    if (!migrate_use_compression()) {
> +        return;
> +    }
> +    quit_thread = false;
> +    thread_count = migrate_compress_threads();
> +    s->compress_thread = g_malloc0(sizeof(QemuThread)
> +        * thread_count);
> +    comp_param = g_malloc0(sizeof(compress_param) * thread_count);

I think g_new0 is better for that.

> +    for (i = 0; i < thread_count; i++) {
> +        qemu_thread_create(s->compress_thread + i, "compress",
> +            do_data_compress, comp_param + i, QEMU_THREAD_JOINABLE);
> +
> +    }
> +}
> +
>  /* Update the xbzrle cache to reflect a page that's been sent as all 0.
>   * The important thing is that a stale (not-yet-0'd) page be replaced
>   * by the new data.
> @@ -643,6 +704,16 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
>      return bytes_sent;
>  }
>  
> +static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
> +        ram_addr_t offset, bool last_stage)
> +{
> +    int bytes_sent = 0;
> +
> +    /* To be done*/
> +
> +    return bytes_sent;
> +}
> +
>  /*
>   * ram_find_and_save_block: Finds a page to send and sends it to f
>   *
> @@ -677,7 +748,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
>                  ram_bulk_stage = false;
>              }
>          } else {
> -            bytes_sent = ram_save_page(f, block, offset, last_stage);
> +            if (migrate_use_compression()) {
> +                bytes_sent = ram_save_compressed_page(f,
> +                        block, offset, last_stage);
> +            } else {
> +                bytes_sent = ram_save_page(f, block, offset, last_stage);
> +            }
>  
>              /* if page is unmodified, continue to the next */
>              if (bytes_sent > 0) {
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index 3cb5ba8..daf6c81 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -49,6 +49,9 @@ struct MigrationState
>      QemuThread thread;
>      QEMUBH *cleanup_bh;
>      QEMUFile *file;
> +    QemuThread *compress_thread;
> +    int compress_thread_count;
> +    int compress_level;

OK, these get replaced later anyway by the parameter stuff.

>  
>      int state;
>      MigrationParams params;
> @@ -107,6 +110,8 @@ bool migration_has_finished(MigrationState *);
>  bool migration_has_failed(MigrationState *);
>  MigrationState *migrate_get_current(void);
>  
> +void migrate_compress_threads_create(MigrationState *s);
> +void migrate_compress_threads_join(MigrationState *s);
>  uint64_t ram_bytes_remaining(void);
>  uint64_t ram_bytes_transferred(void);
>  uint64_t ram_bytes_total(void);
> @@ -156,6 +161,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.c b/migration.c
> index c49a05a..402daae 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -43,6 +43,11 @@ enum {
>  #define BUFFER_DELAY     100
>  #define XFER_LIMIT_RATIO (1000 / BUFFER_DELAY)
>  
> +/* Default compression thread count */
> +#define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
> +/*0: means nocompress, 1: best speed, ... 9: best compress ratio */
> +#define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
> +
>  /* Migration XBZRLE default cache size */
>  #define DEFAULT_MIGRATE_CACHE_SIZE (64 * 1024 * 1024)
>  
> @@ -60,6 +65,8 @@ MigrationState *migrate_get_current(void)
>          .bandwidth_limit = MAX_THROTTLE,
>          .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
>          .mbps = -1,
> +        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> +        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
>      };
>  
>      return &current_migration;
> @@ -302,6 +309,7 @@ static void migrate_fd_cleanup(void *opaque)
>          qemu_thread_join(&s->thread);
>          qemu_mutex_lock_iothread();
>  
> +        migrate_compress_threads_join(s);
>          qemu_fclose(s->file);
>          s->file = NULL;
>      }
> @@ -373,6 +381,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));
> @@ -383,6 +393,8 @@ static MigrationState *migrate_init(const MigrationParams *params)
>             sizeof(enabled_capabilities));
>      s->xbzrle_cache_size = xbzrle_cache_size;
>  
> +    s->compress_level = compress_level;
> +    s->compress_thread_count = compress_thread_count;
>      s->bandwidth_limit = bandwidth_limit;
>      s->state = MIG_STATE_SETUP;
>      trace_migrate_set_state(MIG_STATE_SETUP);
> @@ -555,6 +567,31 @@ bool migrate_zero_blocks(void)
>      return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS];
>  }
>  
> +bool migrate_use_compression(void)
> +{
> +    /* Disable compression before the series of patches 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;
> @@ -695,6 +732,7 @@ void migrate_fd_connect(MigrationState *s)
>      /* Notify before starting migration thread */
>      notifier_list_notify(&migration_state_notifiers, s);
>  
> +    migrate_compress_threads_create(s);
>      qemu_thread_create(&s->thread, "migration", migration_thread, s,
>                         QEMU_THREAD_JOINABLE);
>  }
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 03/13] migration: Add the framework of muti-thread decompression

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c                   | 70 +++++++++++++++++++++++++++++++++++++++++++
>  include/migration/migration.h |  4 +++
>  migration.c                   | 15 ++++++++++
>  3 files changed, 89 insertions(+)
> 
> diff --git a/arch_init.c b/arch_init.c
> index a988ec2..2f1d0c4 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -126,6 +126,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;
> @@ -332,13 +333,26 @@ static uint64_t migration_dirty_pages;
>  static uint32_t last_version;
>  static bool ram_bulk_stage;
>  
> +/* using compressBound() to calculate the buffer size needed to save the
> + * compressed data, to support the maximun TARGET_PAGE_SIZE bytes of

Typo: 'maximun'->'maximum'

> + * data, we need more 15 bytes, use 16 to align the data.
> + */
> +#define COMPRESS_BUF_SIZE (TARGET_PAGE_SIZE + 16)
> +
>  struct compress_param {
>      /* To be done */
>  };
>  typedef struct compress_param compress_param;
>  
> +struct decompress_param {
> +    /* To be done */
> +};
> +typedef struct decompress_param decompress_param;
> +
>  static compress_param *comp_param;
>  static bool quit_thread;
> +static decompress_param *decomp_param;
> +static QemuThread *decompress_threads;
>  
>  static void *do_data_compress(void *opaque)
>  {
> @@ -1124,10 +1138,54 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
>      }
>  }
>  
> +static void *do_data_decompress(void *opaque)
> +{
> +    while (!quit_thread) {
> +        /* To be done */
> +    }
> +    return NULL;
> +}
> +
> +void migrate_decompress_threads_create(int count)
> +{
> +    int i;
> +
> +    decompress_threads = g_malloc0(sizeof(QemuThread) * count);
> +    decomp_param = g_malloc0(sizeof(decompress_param) * count);

Again, g_new0 probably better.

> +    quit_thread = false;
> +    for (i = 0; i < count; i++) {
> +        qemu_thread_create(decompress_threads + i, "decompress",
> +            do_data_decompress, decomp_param + i, QEMU_THREAD_JOINABLE);
> +    }
> +}
> +
> +void migrate_decompress_threads_join(void)
> +{
> +    int i, thread_count;
> +
> +    quit_thread = true;
> +    thread_count = migrate_decompress_threads();
> +    for (i = 0; i < thread_count; i++) {
> +        qemu_thread_join(decompress_threads + i);
> +    }
> +    g_free(decompress_threads);
> +    g_free(decomp_param);
> +    decompress_threads = NULL;
> +    decomp_param = 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;
> +    uint8_t compbuf[COMPRESS_BUF_SIZE];
>  
>      seq_iter++;
>  
> @@ -1201,6 +1259,18 @@ 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("Illegal RAM offset " RAM_ADDR_FMT, addr);
> +                ret = -EINVAL;
> +                break;
> +            }
> +
> +            len = qemu_get_be32(f);
> +            qemu_get_buffer(f, compbuf, len);

I think you need to check 'len' is sensible, if your stream goes wrong it
could end up being a negative or silly value.

> +            decompress_data_with_multi_threads(compbuf, 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 daf6c81..0c4f21c 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -51,6 +51,7 @@ struct MigrationState
>      QEMUFile *file;
>      QemuThread *compress_thread;
>      int compress_thread_count;
> +    int decompress_thread_count;
>      int compress_level;
>  
>      int state;
> @@ -112,6 +113,8 @@ MigrationState *migrate_get_current(void);
>  
>  void migrate_compress_threads_create(MigrationState *s);
>  void migrate_compress_threads_join(MigrationState *s);
> +void migrate_decompress_threads_create(int count);
> +void migrate_decompress_threads_join(void);
>  uint64_t ram_bytes_remaining(void);
>  uint64_t ram_bytes_transferred(void);
>  uint64_t ram_bytes_total(void);
> @@ -164,6 +167,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.c b/migration.c
> index 402daae..082ddb7 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -45,6 +45,7 @@ enum {
>  
>  /* Default compression thread count */
>  #define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
> +#define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
>  /*0: means nocompress, 1: best speed, ... 9: best compress ratio */
>  #define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
>  
> @@ -66,6 +67,7 @@ MigrationState *migrate_get_current(void)
>          .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
>          .mbps = -1,
>          .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> +        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
>          .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
>      };
>  
> @@ -123,10 +125,13 @@ static void process_incoming_migration_co(void *opaque)
>      } else {
>          runstate_set(RUN_STATE_PAUSED);
>      }
> +    migrate_decompress_threads_join();
>  }
>  
>  void process_incoming_migration(QEMUFile *f)
>  {
> +    int thread_count = migrate_decompress_threads();
> +    migrate_decompress_threads_create(thread_count);
>      Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
>      int fd = qemu_get_fd(f);

The style rules need all of the declarations to be kept together;
so the 'migrate_decompress_threads_create' can't go in between.

> @@ -383,6 +388,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));
> @@ -395,6 +401,7 @@ static MigrationState *migrate_init(const MigrationParams *params)
>  
>      s->compress_level = compress_level;
>      s->compress_thread_count = compress_thread_count;
> +    s->decompress_thread_count = decompress_thread_count;
>      s->bandwidth_limit = bandwidth_limit;
>      s->state = MIG_STATE_SETUP;
>      trace_migrate_set_state(MIG_STATE_SETUP);
> @@ -591,6 +598,14 @@ 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)
>  {
> -- 
> 1.8.3.1

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

Re: [Qemu-devel] [v3 04/13] qemu-file: Add tow function will be used in migration

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> migrate_qemu_add_compression_data() compress the data
> and put it to QEMUFile. migrate_qemu_flush() put the
> data in the source QEMUFile to destination QEMUFile.
> 
> The two function can help to do live migration.

Typo in the title 'tow->two' - but perhaps a better title
would be 'Add compression functions to QEMUFile'

> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  include/migration/qemu-file.h |  3 +++
>  qemu-file.c                   | 32 ++++++++++++++++++++++++++++++++
>  2 files changed, 35 insertions(+)
> 
> diff --git a/include/migration/qemu-file.h b/include/migration/qemu-file.h
> index 401676b..d70fb8d 100644
> --- a/include/migration/qemu-file.h
> +++ b/include/migration/qemu-file.h
> @@ -150,6 +150,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);
> +size_t migrate_qemu_add_compression_data(QEMUFile *f,
> +        const uint8_t *p, size_t size, int level);
> +int migrate_qemu_flush(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/qemu-file.c b/qemu-file.c
> index f938e36..2668ad0 100644
> --- a/qemu-file.c
> +++ b/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"
> @@ -993,3 +994,34 @@ QEMUFile *qemu_bufopen(const char *mode, QEMUSizedBuffer *input)
>      }
>      return s->file;
>  }
> +
> +/* compress size bytes of data start at p with specific compression
> + * leve and store the compressed data to the buffer of f.
> + */
> +
> +size_t migrate_qemu_add_compression_data(QEMUFile *f,
> +        const uint8_t *p, size_t size, int level)

It's an odd name, QEMUFile is only used by migration anyway;
maybe qemufile_add_compression_data ?

> +{
> +    size_t blen = IO_BUF_SIZE - f->buf_index - sizeof(int);
> +
> +    if (compress2(f->buf + f->buf_index + sizeof(int), &blen, (Bytef *)p,
> +            size, level) != Z_OK) {
> +        error_report("Compress Failed!");
> +        return 0;
> +    }

What are the 'sizeof(int)'s for?  It's unusual because we normally keep the
format of the stream the same even if one side was a 32bit qemu and the other 64bit.
How do you know that there is enough space for the compress - i.e. what
happens if f->buf_index is too high?

> +    qemu_put_be32(f, blen);
> +    f->buf_index += blen;
> +    return blen + sizeof(int);
> +}
> +
> +int migrate_qemu_flush(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;
> +}

Can you explain a bit more here how you're using the src file; I think
you're using it as kind of a dummy buffer; but it needs to be documented
somewhere.  Again I'm also not sure of the name of this function.

Dave

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

Re: [Qemu-devel] [v3 05/13] arch_init: alloc and free data struct in multi-thread compression

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Define the data structure and varibles used when doing 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>
> ---
>  arch_init.c | 34 +++++++++++++++++++++++++++++++++-
>  1 file changed, 33 insertions(+), 1 deletion(-)
> 
> diff --git a/arch_init.c b/arch_init.c
> index 2f1d0c4..f21a8ea 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -340,16 +340,29 @@ static bool ram_bulk_stage;
>  #define COMPRESS_BUF_SIZE (TARGET_PAGE_SIZE + 16)
>  
>  struct compress_param {
> -    /* To be done */
> +    int state;
> +    QEMUFile *file;
> +    QemuMutex mutex;
> +    QemuCond cond;
> +    RAMBlock *block;
> +    ram_addr_t offset;
>  };
>  typedef struct compress_param compress_param;
>  
> +enum {
> +    DONE,
> +    START,
> +};
> +

Do you really need any more than a 'bool busy' ?

>  struct decompress_param {
>      /* To be done */
>  };
>  typedef struct decompress_param decompress_param;
>  
>  static compress_param *comp_param;
> +static QemuMutex *mutex;
> +static QemuCond *cond;

Those need better names and a comment; If I'm reading it
correctly, this cond is used to wake up the parent thread
when one of the workers has finished it's task?

> +static QEMUFileOps *empty_ops;
>  static bool quit_thread;
>  static decompress_param *decomp_param;
>  static QemuThread *decompress_threads;
> @@ -381,11 +394,22 @@ void migrate_compress_threads_join(MigrationState *s)
>      thread_count = migrate_compress_threads();
>      for (i = 0; i < thread_count; i++) {
>          qemu_thread_join(s->compress_thread + i);
> +        qemu_fclose(comp_param[i].file);
> +        qemu_mutex_destroy(&comp_param[i].mutex);
> +        qemu_cond_destroy(&comp_param[i].cond);
>      }
> +    qemu_mutex_destroy(mutex);
> +    qemu_cond_destroy(cond);
>      g_free(s->compress_thread);
>      g_free(comp_param);
> +    g_free(cond);
> +    g_free(mutex);
> +    g_free(empty_ops);
>      s->compress_thread = NULL;
>      comp_param = NULL;
> +    cond = NULL;
> +    mutex = NULL;
> +    empty_ops = NULL;
>  }
>  
>  void migrate_compress_threads_create(MigrationState *s)
> @@ -400,7 +424,15 @@ void migrate_compress_threads_create(MigrationState *s)
>      s->compress_thread = g_malloc0(sizeof(QemuThread)
>          * thread_count);
>      comp_param = g_malloc0(sizeof(compress_param) * thread_count);
> +    cond = g_malloc0(sizeof(QemuCond));
> +    mutex = g_malloc0(sizeof(QemuMutex));
> +    empty_ops = g_malloc0(sizeof(QEMUFileOps));

Again this needs to go with the explanation of what you're using
the special QEMUFile for; but I don't think anything outside of
QEMUFile should be allocating a QEMUFileOps (It could be static anyway
rather than malloc'd).  I think you could make empty_ops declared
static in qemu-file.c

> +    qemu_cond_init(cond);
> +    qemu_mutex_init(mutex);
>      for (i = 0; i < thread_count; i++) {
> +        comp_param[i].file = qemu_fopen_ops(NULL, empty_ops);
> +        qemu_mutex_init(&comp_param[i].mutex);
> +        qemu_cond_init(&comp_param[i].cond);
>          qemu_thread_create(s->compress_thread + i, "compress",
>              do_data_compress, comp_param + i, QEMU_THREAD_JOINABLE);
>  
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 07/13] migraion: Rewrite the function ram_save_page()

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> We rewrite this function to reuse the code in it
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c | 107 ++++++++++++++++++++++++++++++++++--------------------------
>  1 file changed, 61 insertions(+), 46 deletions(-)

The title would probably be better as 'Split ram_save_page()' - you
don't actually rewrite the code that much.
Note the important comment below.

> diff --git a/arch_init.c b/arch_init.c
> index 71cc756..0a575ed 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -596,6 +596,63 @@ static void migration_bitmap_sync_range(ram_addr_t start, ram_addr_t length)
>      }
>  }
>  
> +static int save_zero_and_xbzrle_page(QEMUFile *f, RAMBlock* block,
> +        ram_addr_t offset, bool last_stage, bool *send_async)
> +{
> +    int bytes_sent;
> +    int cont;
> +    ram_addr_t current_addr;
> +    MemoryRegion *mr = block->mr;
> +    uint8_t *p;
> +    int ret;
> +
> +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> +
> +    p = memory_region_get_ram_ptr(mr) + offset;
> +
> +    /* In doubt sent page as normal */
> +    bytes_sent = -1;
> +    ret = ram_control_save_page(f, block->offset,
> +                           offset, TARGET_PAGE_SIZE, &bytes_sent);
> +
> +    XBZRLE_cache_lock();
> +
> +    current_addr = block->offset + offset;
> +    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
> +        if (ret != RAM_SAVE_CONTROL_DELAYED) {
> +            if (bytes_sent > 0) {
> +                acct_info.norm_pages++;
> +            } else if (bytes_sent == 0) {
> +                acct_info.dup_pages++;
> +            }
> +        }
> +    } else if (is_zero_range(p, TARGET_PAGE_SIZE)) {
> +        acct_info.dup_pages++;
> +        bytes_sent = save_block_hdr(f, block, offset, cont,
> +                                    RAM_SAVE_FLAG_COMPRESS);
> +        qemu_put_byte(f, 0);
> +        bytes_sent++;
> +        /* Must let xbzrle know, otherwise a previous (now 0'd) cached
> +         * page would be stale
> +         */
> +        xbzrle_cache_zero_page(current_addr);
> +    } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
> +        bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
> +                                      offset, cont, last_stage);
> +        if (!last_stage) {
> +            /* Can't send this cached data async, since the cache page
> +             * might get updated before it gets to the wire
> +             */
> +            if (send_async != NULL) {
> +                *send_async = false;
> +            }
> +        }
> +    }
> +
> +    XBZRLE_cache_unlock();
> +
> +    return bytes_sent;
> +}
>  
>  /* Needs iothread lock! */
>  /* Fix me: there are too many global variables used in migration process. */
> @@ -691,55 +748,15 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
>  {
>      int bytes_sent;
>      int cont;
> -    ram_addr_t current_addr;
>      MemoryRegion *mr = block->mr;
>      uint8_t *p;
> -    int ret;
>      bool send_async = true;
>  
> -    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> -
> -    p = memory_region_get_ram_ptr(mr) + offset;
> -
> -    /* In doubt sent page as normal */
> -    bytes_sent = -1;
> -    ret = ram_control_save_page(f, block->offset,
> -                           offset, TARGET_PAGE_SIZE, &bytes_sent);
> -
> -    XBZRLE_cache_lock();
> -
> -    current_addr = block->offset + offset;
> -    if (ret != RAM_SAVE_CONTROL_NOT_SUPP) {
> -        if (ret != RAM_SAVE_CONTROL_DELAYED) {
> -            if (bytes_sent > 0) {
> -                acct_info.norm_pages++;
> -            } else if (bytes_sent == 0) {
> -                acct_info.dup_pages++;
> -            }
> -        }
> -    } else if (is_zero_range(p, TARGET_PAGE_SIZE)) {
> -        acct_info.dup_pages++;
> -        bytes_sent = save_block_hdr(f, block, offset, cont,
> -                                    RAM_SAVE_FLAG_COMPRESS);
> -        qemu_put_byte(f, 0);
> -        bytes_sent++;
> -        /* Must let xbzrle know, otherwise a previous (now 0'd) cached
> -         * page would be stale
> -         */
> -        xbzrle_cache_zero_page(current_addr);
> -    } else if (!ram_bulk_stage && migrate_use_xbzrle()) {
> -        bytes_sent = save_xbzrle_page(f, &p, current_addr, block,
> -                                      offset, cont, last_stage);
> -        if (!last_stage) {
> -            /* Can't send this cached data async, since the cache page
> -             * might get updated before it gets to the wire
> -             */
> -            send_async = false;
> -        }
> -    }
> -
> -    /* XBZRLE overflow or normal page */
> +    bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
> +            last_stage, &send_async);
>      if (bytes_sent == -1) {
> +        cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> +        p = memory_region_get_ram_ptr(mr) + offset;

I think this breaks XBZRLE; the 'p' pointer is updated by save_xbzrle_page when it
copies the page into the cache; when that happens ram_save_page must use that cache copy
rather than the page in main memory; you're recalculating p.
See the commit 1534ee93 'XBZRLE: Fix one XBZRLE corruption issues'

>          bytes_sent = save_block_hdr(f, block, offset, cont, RAM_SAVE_FLAG_PAGE);
>          if (send_async) {
>              qemu_put_buffer_async(f, p, TARGET_PAGE_SIZE);
> @@ -750,8 +767,6 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
>          acct_info.norm_pages++;
>      }
>  
> -    XBZRLE_cache_unlock();
> -
>      return bytes_sent;
>  }
>  
> -- 
> 1.8.3.1

Dave

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

Re: [Qemu-devel] [v3 08/13] migration: Add the core code of multi-thread compresion

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> At this point, multiple thread compression can't co-work with xbzrle.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c | 164 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++---
>  1 file changed, 157 insertions(+), 7 deletions(-)
> 
> diff --git a/arch_init.c b/arch_init.c
> index 0a575ed..4109ad7 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -369,23 +369,43 @@ static QemuMutex *mutex;
>  static QemuCond *cond;
>  static QEMUFileOps *empty_ops;
>  static bool quit_thread;
> +static int one_byte_count;
>  static decompress_param *decomp_param;
>  static QemuThread *decompress_threads;
>  
> +static int do_compress_ram_page(compress_param *param);
> +
>  static void *do_data_compress(void *opaque)
>  {
> +    compress_param *param = opaque;
>      while (!quit_thread) {
> -
> -    /* To be done */
> -
> +        qemu_mutex_lock(&param->mutex);
> +        while (param->state != START) {
> +            qemu_cond_wait(&param->cond, &param->mutex);
> +            if (quit_thread) {
> +                break;
> +            }
> +            do_compress_ram_page(param);
> +            qemu_mutex_lock(mutex);
> +            param->state = DONE;
> +            qemu_cond_signal(cond);
> +            qemu_mutex_unlock(mutex);
> +        }
> +        qemu_mutex_unlock(&param->mutex);
>      }
> +
>      return NULL;
>  }
>  
>  static inline void terminate_compression_threads(void)
>  {
> +    int idx, thread_count;
> +
> +    thread_count = migrate_compress_threads();
>      quit_thread = true;
> -    /* To be done */
> +    for (idx = 0; idx < thread_count; idx++) {
> +        qemu_cond_signal(&comp_param[idx].cond);
> +    }
>  }
>  
>  void migrate_compress_threads_join(MigrationState *s)
> @@ -770,13 +790,142 @@ static int ram_save_page(QEMUFile *f, RAMBlock* block, ram_addr_t offset,
>      return bytes_sent;
>  }
>  
> +static int do_compress_ram_page(compress_param *param)
> +{
> +    int bytes_sent;
> +    int blen = COMPRESS_BUF_SIZE;
> +    int cont;
> +    uint8_t *p;
> +    RAMBlock *block = param->block;
> +    ram_addr_t offset = param->offset;
> +
> +    cont = (block == last_sent_block) ? RAM_SAVE_FLAG_CONTINUE : 0;
> +    p = memory_region_get_ram_ptr(block->mr) + offset;
> +
> +    bytes_sent = save_block_hdr(param->file, block,
> +            offset, cont, RAM_SAVE_FLAG_COMPRESS_PAGE);
> +    blen = migrate_qemu_add_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(compress_param *param)
> +{
> +    qemu_mutex_lock(&param->mutex);
> +    param->state = START;
> +    qemu_cond_signal(&param->cond);
> +    qemu_mutex_unlock(&param->mutex);
> +}
> +
> +
> +static uint64_t bytes_transferred;
> +
> +static void flush_compressed_data(QEMUFile *f)
> +{
> +    int idx, len, thread_count;
> +
> +    if (!migrate_use_compression()) {
> +        return;
> +    }
> +    thread_count = migrate_compress_threads();
> +    for (idx = 0; idx < thread_count; idx++) {
> +        if (comp_param[idx].state != DONE) {
> +            qemu_mutex_lock(mutex);
> +            while (comp_param[idx].state != DONE) {
> +                qemu_cond_wait(cond, mutex);
> +            }
> +            qemu_mutex_unlock(mutex);
> +        }
> +        len = migrate_qemu_flush(f, comp_param[idx].file);
> +        bytes_transferred += len;
> +    }
> +    if ((one_byte_count > 0) && (bytes_transferred > one_byte_count)) {
> +        bytes_transferred -= one_byte_count;
> +        one_byte_count = 0;
> +    }
> +}
> +
> +static inline void set_compress_params(compress_param *param,
> +    RAMBlock *block, ram_addr_t offset)
> +{
> +    param->block = block;
> +    param->offset = offset;
> +}
> +
> +
> +static int compress_page_with_multi_thread(QEMUFile *f,
> +        RAMBlock *block, ram_addr_t offset)
> +{
> +    int idx, thread_count, bytes_sent = 0;
> +
> +    thread_count = migrate_compress_threads();
> +    qemu_mutex_lock(mutex);
> +    while (true) {
> +        for (idx = 0; idx < thread_count; idx++) {
> +            if (comp_param[idx].state == DONE) {
> +                bytes_sent = migrate_qemu_flush(f, comp_param[idx].file);
> +                set_compress_params(&comp_param[idx],
> +                        block, offset);
> +                start_compression(&comp_param[idx]);
> +                if (bytes_sent == 0) {
> +                    /* set bytes_sent to 1 in this case to prevent migration
> +                     * from terminating, this 1 byte whill be added to
> +                     * bytes_transferred later, minus 1 to keep the
> +                     * bytes_transferred accurate */
> +                    bytes_sent = 1;
> +                    if (bytes_transferred <= 0) {
> +                        one_byte_count++;
> +                    } else {
> +                        bytes_transferred -= 1;
> +                    }
> +                }
> +                break;
> +            }
> +        }
> +        if (bytes_sent > 0) {
> +            break;
> +        } else {
> +            qemu_cond_wait(cond, mutex);
> +        }
> +    }
> +    qemu_mutex_unlock(mutex);
> +    return bytes_sent;
> +}
> +
>  static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
>          ram_addr_t offset, bool last_stage)
>  {
>      int bytes_sent = 0;
>  
> -    /* To be done*/
> -
> +    /* 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.

Why?  Is this just because of the 'cont' flag used to avoid sending the
block names again?

Dave

> +     */
> +    if (block != last_sent_block) {
> +        flush_compressed_data(f);
> +        bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
> +                last_stage, NULL);
> +        if (bytes_sent == -1) {
> +            set_compress_params(&comp_param[0], block, offset);
> +            /* Use the qemu thread to compress the data to make sure the
> +             * first page is sent out before other pages
> +             */
> +            bytes_sent = do_compress_ram_page(&comp_param[0]);
> +            if (bytes_sent > 0) {
> +                migrate_qemu_flush(f, comp_param[0].file);
> +            }
> +        }
> +    } else {
> +        bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
> +                last_stage, NULL);
> +        if (bytes_sent == -1) {
> +            bytes_sent = compress_page_with_multi_thread(f, block, offset);
> +        }
> +    }
>      return bytes_sent;
>  }
>  
> @@ -834,7 +983,6 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
>      return bytes_sent;
>  }
>  
> -static uint64_t bytes_transferred;
>  
>  void acct_update_position(QEMUFile *f, size_t size, bool zero)
>  {
> @@ -1043,6 +1191,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
>          i++;
>      }
>  
> +    flush_compressed_data(f);
>      qemu_mutex_unlock_ramlist();
>  
>      /*
> @@ -1089,6 +1238,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
>          bytes_transferred += bytes_sent;
>      }
>  
> +    flush_compressed_data(f);
>      ram_control_after_iterate(f, RAM_CONTROL_FINISH);
>      migration_end();
>  
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

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

[Dr. David Alan Gilbert]

* 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>
> ---
>  arch_init.c | 13 +++++++++++--
>  1 file changed, 11 insertions(+), 2 deletions(-)
> 
> diff --git a/arch_init.c b/arch_init.c
> index 4109ad7..14bc486 100644
> --- a/arch_init.c
> +++ b/arch_init.c
> @@ -905,8 +905,11 @@ static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,
>       * block, and all the pages in last block should have been sent
>       * out, keeping this order is important.
>       */
> -    if (block != last_sent_block) {
> -        flush_compressed_data(f);
> +    if ((!ram_bulk_stage && migrate_use_xbzrle()) ||
> +                block != last_sent_block) {
> +        if (block != last_sent_block) {
> +            flush_compressed_data(f);
> +        }
>          bytes_sent = save_zero_and_xbzrle_page(f, block, offset,
>                  last_stage, NULL);
>          if (bytes_sent == -1) {
> @@ -961,6 +964,12 @@ static int ram_find_and_save_block(QEMUFile *f, bool last_stage)
>                  block = QTAILQ_FIRST(&ram_list.blocks);
>                  complete_round = true;
>                  ram_bulk_stage = false;
> +                if (migrate_use_xbzrle()) {
> +                    /* if xbzrle is on, we terminate the compression thread
> +                     * at this point, there is no benefit from muti-thead */

Typo: 'muti-thead' -> 'multi-thread'

Dave

> +                    flush_compressed_data(f);
> +                    terminate_compression_threads();
> +                }
>              }
>          } else {
>              if (migrate_use_compression()) {
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 10/13] migration: Add the core code of multi-thread decompression

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c | 48 ++++++++++++++++++++++++++++++++++++++++++++++--
>  1 file changed, 46 insertions(+), 2 deletions(-)
> 
> diff --git a/arch_init.c b/arch_init.c
> index 14bc486..7103f4f 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>
> @@ -820,6 +821,14 @@ static inline void start_compression(compress_param *param)
>      qemu_mutex_unlock(&param->mutex);
>  }
>  
> +static inline void start_decompression(decompress_param *param)
> +{
> +    qemu_mutex_lock(&param->mutex);
> +    param->state = START;
> +    qemu_cond_signal(&param->cond);
> +    qemu_mutex_unlock(&param->mutex);
> +}
> +
>  
>  static uint64_t bytes_transferred;
>  
> @@ -1351,8 +1360,24 @@ void ram_handle_compressed(void *host, uint8_t ch, uint64_t size)
>  
>  static void *do_data_decompress(void *opaque)
>  {
> +    decompress_param *param = opaque;
>      while (!quit_thread) {
> -        /* To be done */
> +        qemu_mutex_lock(&param->mutex);
> +        while (param->state != START) {
> +            qemu_cond_wait(&param->cond, &param->mutex);
> +            if (quit_thread) {
> +                break;
> +            }
> +            size_t pagesize = TARGET_PAGE_SIZE;
> +            /* uncompress() will return failed in some case,
> +             * especially when the page is dirted when doing
> +             * the compression, ignore the return value because
> +             * the dirty page will be retransferred. */
> +            uncompress((Bytef *)param->des, &pagesize,
> +                    (const Bytef *)param->compbuf, param->len);

That's kind of a scary comment!  It looks like 'uncompress' is supposed
to be safe, so shouldn't damage any other data; it's worrying me
might not find real problems though.

However,

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

> +            param->state = DONE;
> +        }
> +        qemu_mutex_unlock(&param->mutex);
>      }
>      return NULL;
>  }
> @@ -1379,6 +1404,9 @@ void migrate_decompress_threads_join(void)
>      quit_thread = true;
>      thread_count = migrate_decompress_threads();
>      for (i = 0; i < thread_count; i++) {
> +        qemu_cond_signal(&decomp_param[i].cond);
> +    }
> +    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);
> @@ -1392,7 +1420,23 @@ void migrate_decompress_threads_join(void)
>  static void decompress_data_with_multi_threads(uint8_t *compbuf,
>          void *host, int len)
>  {
> -    /* To be done */
> +    int idx, thread_count;
> +
> +    thread_count = migrate_decompress_threads();
> +    while (true) {
> +        for (idx = 0; idx < thread_count; idx++) {
> +            if (decomp_param[idx].state == DONE) {
> +                memcpy(decomp_param[idx].compbuf, compbuf, len);
> +                decomp_param[idx].des = host;
> +                decomp_param[idx].len = len;
> +                start_decompression(&decomp_param[idx]);
> +                break;
> +            }
> +        }
> +        if (idx < thread_count) {
> +            break;
> +        }
> +    }
>  }
>  
>  static int ram_load(QEMUFile *f, void *opaque, int version_id)
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 11/13] migration: Add interface to control compression

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> The multiple compression threads can be turned on/off through
> qmp and hmp interface when doing live migration.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  migration.c      | 7 +++++--
>  qapi-schema.json | 6 +++++-
>  2 files changed, 10 insertions(+), 3 deletions(-)
> 

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

> diff --git a/migration.c b/migration.c
> index 082ddb7..9d1613d 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -576,8 +576,11 @@ bool migrate_zero_blocks(void)
>  
>  bool migrate_use_compression(void)
>  {
> -    /* Disable compression before the series of patches 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 9ffdcf8..d371af3 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -491,13 +491,17 @@
>  #          to enable the capability on the source VM. The feature is disabled by
>  #          default. (since 1.6)
>  #
> +# @compress: Using the multiple compression threads to accelerate live migration.
> +#          This feature can help to reduce the migration traffic, by sending
> +#          compressed pages. The feature is disabled by default. (since 2.3)
> +#
>  # @auto-converge: If enabled, QEMU will automatically throttle down the guest
>  #          to speed up convergence of RAM migration. (since 1.6)
>  #
>  # Since: 1.2
>  ##
>  { 'enum': 'MigrationCapability',
> -  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks'] }
> +  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks', 'compress'] }
>  
>  ##
>  # @MigrationCapabilityStatus
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 12/13] migration: Add command to set migration parameter

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Add the qmp and hmp commands to tune the parameters used in live
> migration.

If I understand correctly on the destination side we need to set the number of
decompression threads very early on an incoming migration - I'm not
clear how early that needs to be - especially if you're using fd: so it's
not waiting for a connect ?

Eric: How would libvirt do that?

> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  hmp-commands.hx               | 15 ++++++++++
>  hmp.c                         | 32 +++++++++++++++++++++
>  hmp.h                         |  3 ++
>  include/migration/migration.h |  4 +--
>  migration.c                   | 66 +++++++++++++++++++++++++++++++++++--------
>  monitor.c                     | 18 ++++++++++++
>  qapi-schema.json              | 44 +++++++++++++++++++++++++++++
>  qmp-commands.hx               | 23 +++++++++++++++
>  8 files changed, 190 insertions(+), 15 deletions(-)
> 
> diff --git a/hmp-commands.hx b/hmp-commands.hx
> index e37bc8b..535b5ba 100644
> --- a/hmp-commands.hx
> +++ b/hmp-commands.hx
> @@ -985,6 +985,21 @@ Enable/Disable the usage of a capability @var{capability} for migration.
>  ETEXI
>  
>      {
> +        .name       = "migrate_set_parameter",
> +        .args_type  = "parameter:s,value:i",
> +        .params     = "parameter value",
> +        .help       = "Set the parameter for migration",
> +        .mhandler.cmd = hmp_migrate_set_parameter,
> +        .command_completion = migrate_set_parameter_completion,
> +    },
> +
> +STEXI
> +@item migrate_set_parameter @var{parameter} @var{value}
> +@findex migrate_set_parameter
> +Set the parameter @var{parameter} for migration.
> +ETEXI
> +
> +    {
>          .name       = "client_migrate_info",
>          .args_type  = "protocol:s,hostname:s,port:i?,tls-port:i?,cert-subject:s?",
>          .params     = "protocol hostname port tls-port cert-subject",
> diff --git a/hmp.c b/hmp.c
> index 63d7686..965c037 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -1079,6 +1079,38 @@ void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict)
>      }
>  }
>  
> +void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
> +{
> +    const char *param = qdict_get_str(qdict, "parameter");
> +    int value = qdict_get_int(qdict, "value");
> +    Error *err = NULL;
> +    MigrationParameterStatusList *params = g_malloc0(sizeof(*params));
> +    int i;
> +
> +    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +        if (strcmp(param, MigrationParameter_lookup[i]) == 0) {
> +            params->value = g_malloc0(sizeof(*params->value));
> +            params->value->parameter = i;
> +            params->value->value = value;
> +            params->next = NULL;
> +            qmp_migrate_set_parameters(params, &err);
> +            break;
> +        }
> +    }
> +
> +    if (i == MIGRATION_PARAMETER_MAX) {
> +        error_set(&err, QERR_INVALID_PARAMETER, param);
> +    }
> +
> +    qapi_free_MigrationParameterStatusList(params);
> +
> +    if (err) {
> +        monitor_printf(mon, "migrate_set_parameter: %s\n",
> +                       error_get_pretty(err));
> +        error_free(err);
> +    }
> +}
> +
>  void hmp_set_password(Monitor *mon, const QDict *qdict)
>  {
>      const char *protocol  = qdict_get_str(qdict, "protocol");
> diff --git a/hmp.h b/hmp.h
> index 4bb5dca..bd1b203 100644
> --- a/hmp.h
> +++ b/hmp.h
> @@ -63,6 +63,7 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict);
> +void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_cache_size(Monitor *mon, const QDict *qdict);
>  void hmp_set_password(Monitor *mon, const QDict *qdict);
>  void hmp_expire_password(Monitor *mon, const QDict *qdict);
> @@ -111,6 +112,8 @@ void watchdog_action_completion(ReadLineState *rs, int nb_args,
>                                  const char *str);
>  void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
>                                         const char *str);
> +void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
> +                                       const char *str);
>  void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str);
>  void host_net_remove_completion(ReadLineState *rs, int nb_args,
>                                  const char *str);
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index 0c4f21c..8e09b42 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -50,9 +50,7 @@ struct MigrationState
>      QEMUBH *cleanup_bh;
>      QEMUFile *file;
>      QemuThread *compress_thread;
> -    int compress_thread_count;
> -    int decompress_thread_count;
> -    int compress_level;
> +    int parameters[MIGRATION_PARAMETER_MAX];
>  
>      int state;
>      MigrationParams params;
> diff --git a/migration.c b/migration.c
> index 9d1613d..d3d377e 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -66,9 +66,12 @@ MigrationState *migrate_get_current(void)
>          .bandwidth_limit = MAX_THROTTLE,
>          .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
>          .mbps = -1,
> -        .compress_thread_count = DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> -        .decompress_thread_count = DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
> -        .compress_level = DEFAULT_MIGRATE_COMPRESS_LEVEL,
> +        .parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] =
> +                DEFAULT_MIGRATE_COMPRESS_LEVEL,
> +        .parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
> +                DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT,
> +        .parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
> +                DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT,
>      };
>  
>      return &current_migration;
> @@ -292,6 +295,41 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
>      }
>  }
>  
> +void qmp_migrate_set_parameters(MigrationParameterStatusList *params,
> +                                  Error **errp)
> +{
> +    MigrationState *s = migrate_get_current();
> +    MigrationParameterStatusList *p;
> +
> +    for (p = params; p; p = p->next) {
> +        switch (p->value->parameter) {
> +        case MIGRATION_PARAMETER_COMPRESS_LEVEL:
> +            if (p->value->value < 0 || p->value->value > 9) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
> +                    "is invalied, it should be in the rang of 0 to 9");

Typo: 'rang'

> +                return;
> +            }
> +            break;
> +        case MIGRATION_PARAMETER_COMPRESS_THREADS:
> +        case MIGRATION_PARAMETER_DECOMPRESS_THREADS:
> +            if (s->state == MIG_STATE_ACTIVE || s->state == MIG_STATE_SETUP) {
> +                error_set(errp, QERR_MIGRATION_ACTIVE);
> +                return;
> +            }
> +            if (p->value->value < 1 || p->value->value > 255) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE,
> +                    "(de)compress_threads",
> +                    "is invalied, it should be in the rang of 1 to 255");

Typo: 'rang'

> +                return;
> +            }
> +            break;
> +        default:
> +           return;
> +        }
> +        s->parameters[p->value->parameter] = p->value->value;
> +    }
> +}

Do you need the same check here that is in qmp_migrate_set_capabilities;
to stop someone changing a parameter (e.g. the number of threads) while
it's running.

Dave

> +
>  /* shared migration helpers */
>  
>  static void migrate_set_state(MigrationState *s, int old_state, int new_state)
> @@ -386,9 +424,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));
> @@ -399,9 +439,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
>             sizeof(enabled_capabilities));
>      s->xbzrle_cache_size = xbzrle_cache_size;
>  
> -    s->compress_level = compress_level;
> -    s->compress_thread_count = compress_thread_count;
> -    s->decompress_thread_count = decompress_thread_count;
> +    s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
> +    s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
> +               compress_thread_count;
> +    s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
> +               decompress_thread_count;
>      s->bandwidth_limit = bandwidth_limit;
>      s->state = MIG_STATE_SETUP;
>      trace_migrate_set_state(MIG_STATE_SETUP);
> @@ -589,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)
> @@ -598,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)
> @@ -607,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/monitor.c b/monitor.c
> index f1031a1..4cf62b6 100644
> --- a/monitor.c
> +++ b/monitor.c
> @@ -4544,6 +4544,24 @@ void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
>      }
>  }
>  
> +void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
> +                                       const char *str)
> +{
> +    size_t len;
> +
> +    len = strlen(str);
> +    readline_set_completion_index(rs, len);
> +    if (nb_args == 2) {
> +        int i;
> +        for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +            const char *name = MigrationParameter_lookup[i];
> +            if (!strncmp(str, name, len)) {
> +                readline_add_completion(rs, name);
> +            }
> +        }
> +    }
> +}
> +
>  void host_net_add_completion(ReadLineState *rs, int nb_args, const char *str)
>  {
>      int i;
> diff --git a/qapi-schema.json b/qapi-schema.json
> index d371af3..2caeccc 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -540,6 +540,50 @@
>  ##
>  { '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.
> +#
> +# Since: 2.3
> +##
> +{ 'enum': 'MigrationParameter',
> +  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
> +##
> +# @MigrationParameterStatus
> +#
> +# Migration parameter information
> +#
> +# @parameter: parameter enum
> +#
> +# @value: parameter value int
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterStatus',
> +  'data': { 'parameter' : 'MigrationParameter', 'value' : 'int' } }
> +##
> +# @migrate-set-parameters
> +#
> +# Set the following migration parameters (like compress-level)
> +#
> +# @parameters: json array of parameter modifications to make
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'migrate-set-parameters',
> +  'data': { 'parameters': ['MigrationParameterStatus'] } }
> +##
>  ##
>  # @MouseInfo:
>  #
> diff --git a/qmp-commands.hx b/qmp-commands.hx
> index 718dd92..59d2643 100644
> --- a/qmp-commands.hx
> +++ b/qmp-commands.hx
> @@ -3225,6 +3225,29 @@ EQMP
>      },
>  
>  SQMP
> +migrate-set-parameters
> +----------------------
> +
> +Set migration parameters
> +
> +- "compress-leve": multiple compression thread support

Typo: 'compress-leve'

> +
> +Arguments:
> +
> +Example:
> +
> +-> { "execute": "migrate-set-parameters" , "arguments":
> +     { "parameters": [ { "parameter": "compress-level", "value": 1 } ] } }
> +
> +EQMP
> +
> +    {
> +        .name       = "migrate-set-parameters",
> +        .args_type  = "parameters:O",
> +        .params     = "parameter:s,value:i",
> +	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
> +    },
> +SQMP
>  query-balloon
>  -------------
>  
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [v3 13/13] migration: Add command to query migration parameter

[Dr. David Alan Gilbert]

* Liang Li (liang.z.li@intel.com) wrote:
> Add the qmp and hmp commands to query the parameters used in live
> migration.

Eric: I'm OK with this,  but since it's interface stuff, I thought
it best to let you check.

Dave

> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  hmp-commands.hx  |  2 ++
>  hmp.c            | 19 +++++++++++++++++++
>  hmp.h            |  1 +
>  migration.c      | 25 +++++++++++++++++++++++++
>  monitor.c        |  7 +++++++
>  qapi-schema.json | 10 ++++++++++
>  qmp-commands.hx  | 24 ++++++++++++++++++++++++
>  7 files changed, 88 insertions(+)
> 
> diff --git a/hmp-commands.hx b/hmp-commands.hx
> index 535b5ba..ed0c06a 100644
> --- a/hmp-commands.hx
> +++ b/hmp-commands.hx
> @@ -1779,6 +1779,8 @@ show user network stack connection states
>  show migration status
>  @item info migrate_capabilities
>  show current migration capabilities
> +@item info migrate_parameters
> +show current migration parameters
>  @item info migrate_cache_size
>  show current migration XBZRLE cache size
>  @item info balloon
> diff --git a/hmp.c b/hmp.c
> index 965c037..b321b15 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -246,6 +246,25 @@ void hmp_info_migrate_capabilities(Monitor *mon, const QDict *qdict)
>      qapi_free_MigrationCapabilityStatusList(caps);
>  }
>  
> +void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
> +{
> +    MigrationParameterStatusList *params, *p;
> +
> +    params = qmp_query_migrate_parameters(NULL);
> +
> +    if (params) {
> +        monitor_printf(mon, "parameters: ");
> +        for (p = params; p; p = p->next) {
> +            monitor_printf(mon, "%s: %d ",
> +                           MigrationParameter_lookup[p->value->parameter],
> +                           (int)p->value->value);
> +        }
> +        monitor_printf(mon, "\n");
> +    }
> +
> +    qapi_free_MigrationParameterStatusList(params);
> +}
> +
>  void hmp_info_migrate_cache_size(Monitor *mon, const QDict *qdict)
>  {
>      monitor_printf(mon, "xbzrel cache size: %" PRId64 " kbytes\n",
> diff --git a/hmp.h b/hmp.h
> index bd1b203..471417c 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);
> diff --git a/migration.c b/migration.c
> index d3d377e..f87aba3 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -179,6 +179,31 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
>      return head;
>  }
>  
> +MigrationParameterStatusList *qmp_query_migrate_parameters(Error **errp)
> +{
> +    MigrationParameterStatusList *head = NULL;
> +    MigrationParameterStatusList *params;
> +    MigrationState *s = migrate_get_current();
> +    int i;
> +
> +    params = NULL; /* silence compiler warning */
> +    for (i = 0; i < MIGRATION_PARAMETER_MAX; i++) {
> +        if (head == NULL) {
> +            head = g_malloc0(sizeof(*params));
> +            params = head;
> +        } else {
> +            params->next = g_malloc0(sizeof(*params));
> +            params = params->next;
> +        }
> +        params->value =
> +            g_malloc(sizeof(*params->value));
> +        params->value->parameter = i;
> +        params->value->value = s->parameters[i];
> +    }
> +
> +    return head;
> +}
> +
>  static void get_xbzrle_cache_stats(MigrationInfo *info)
>  {
>      if (migrate_use_xbzrle()) {
> diff --git a/monitor.c b/monitor.c
> index 4cf62b6..6ab269f 100644
> --- a/monitor.c
> +++ b/monitor.c
> @@ -2862,6 +2862,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     = "",
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 2caeccc..ccdb6b2 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -584,6 +584,16 @@
>  { 'command': 'migrate-set-parameters',
>    'data': { 'parameters': ['MigrationParameterStatus'] } }
>  ##
> +# @query-migrate-parameters
> +#
> +# Returns information about the current migration parameters status
> +#
> +# Returns: @MigrationParametersStatus
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'query-migrate-parameters', 'returns':   ['MigrationParameterStatus']}
> +##
>  ##
>  # @MouseInfo:
>  #
> diff --git a/qmp-commands.hx b/qmp-commands.hx
> index 59d2643..986eb95 100644
> --- a/qmp-commands.hx
> +++ b/qmp-commands.hx
> @@ -3248,6 +3248,30 @@ EQMP
>  	.mhandler.cmd_new = qmp_marshal_input_migrate_set_parameters,
>      },
>  SQMP
> +query-migrate-parameters
> +------------------------
> +
> +Query current migration parameters
> +
> +- "parameters": migration parameters value
> +         - "compress-level" : compress level value (json-int)
> +
> +Arguments:
> +
> +Example:
> +
> +-> { "execute": "query-migrate-parameters" }
> +<- { "return": [ { "value": 1, "parameter": "compress-level" } ] }
> +
> +EQMP
> +
> +    {
> +        .name       = "query-migrate-parameters",
> +        .args_type  = "",
> +        .mhandler.cmd_new = qmp_marshal_input_query_migrate_parameters,
> +    },
> +
> +SQMP
>  query-balloon
>  -------------
>  
> -- 
> 1.8.3.1
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

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

[Eric Blake]

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

On 12/11/2014 06:28 PM, Liang Li wrote:
> Give some details about the multiple compression threads 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>
> ---
>  docs/multi-thread-compression.txt | 141 ++++++++++++++++++++++++++++++++++++++
>  1 file changed, 141 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..3bbc641
> --- /dev/null
> +++ b/docs/multi-thread-compression.txt
> @@ -0,0 +1,141 @@
> +Use multiple thread (de)compression in live migration
> +======================================================
> +Copyright (C) 2014 Intel Corporation

You may want to include 2015 now.


> +Introduction
> +============
> +Instead of sending the guest memory directly, this solution will
> +compress the ram page before sending; after receiving, the data will

s/ram/RAM/

> +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 migration time can also be reduced about
> +70% in a typical case.

Of course, that's all dependent on memory having compressible contents.
 An application that reads /dev/hwrng into a large amount of memory will
not be that compressible :)


> +
> +When to use the multiple thread compression in live migration
> +==============================================================

off-by-one on the divider length


> +Usage
> +=====
> +1. Verify both the source and destination QEMU are able
> +to support the multiple thread compression migration:
> +    {qemu} info_migrate_capablilites

s/capablilites/capabilities/

> +    {qemu} ... compress: off ...
> +
> +2. Activate compression on the souce:

s/souce/source/

> +    {qemu} migrate_set_capability compress on
> +
> +3. Set the compression thread count on source:
> +    {qemu} migrate_set_paramter compress_threads 12

s/paramter/parameter/


> +
> +The following is the default setting:

s/is...setting/are...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
> +setting is not appropriate.

s/setting is/settings are/

-- 
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] [v3 11/13] migration: Add interface to control compression

[Eric Blake]

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

On 12/11/2014 06:29 PM, Liang Li wrote:
> The multiple compression threads can be turned on/off through
> qmp and hmp interface when doing live migration.
> 
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---

> +++ b/qapi-schema.json
> @@ -491,13 +491,17 @@
>  #          to enable the capability on the source VM. The feature is disabled by
>  #          default. (since 1.6)
>  #
> +# @compress: Using the multiple compression threads to accelerate live migration.

s/Using the/Use/

> +  'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks', 'compress'] }

Long line.  Please wrap to keep things under 80 columns.

-- 
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] [v3 12/13] migration: Add command to set migration parameter

[Eric Blake]

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

On 12/11/2014 06:29 PM, Liang Li wrote:
> Add the qmp and hmp commands to tune 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               | 15 ++++++++++
>  hmp.c                         | 32 +++++++++++++++++++++
>  hmp.h                         |  3 ++
>  include/migration/migration.h |  4 +--
>  migration.c                   | 66 +++++++++++++++++++++++++++++++++++--------
>  monitor.c                     | 18 ++++++++++++
>  qapi-schema.json              | 44 +++++++++++++++++++++++++++++
>  qmp-commands.hx               | 23 +++++++++++++++
>  8 files changed, 190 insertions(+), 15 deletions(-)
> 


> +++ b/hmp.h
> @@ -63,6 +63,7 @@ void hmp_migrate_cancel(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_downtime(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_speed(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_capability(Monitor *mon, const QDict *qdict);
> +void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict);
>  void hmp_migrate_set_cache_size(Monitor *mon, const QDict *qdict);
>  void hmp_set_password(Monitor *mon, const QDict *qdict);
>  void hmp_expire_password(Monitor *mon, const QDict *qdict);
> @@ -111,6 +112,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);

Off-by-one on indentation.


> @@ -292,6 +295,41 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
>      }
>  }
>  
> +void qmp_migrate_set_parameters(MigrationParameterStatusList *params,
> +                                  Error **errp)

Indentation is off.

> +{
> +    MigrationState *s = migrate_get_current();
> +    MigrationParameterStatusList *p;
> +
> +    for (p = params; p; p = p->next) {
> +        switch (p->value->parameter) {
> +        case MIGRATION_PARAMETER_COMPRESS_LEVEL:
> +            if (p->value->value < 0 || p->value->value > 9) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
> +                    "is invalied, it should be in the rang of 0 to 9");

s/invalied/invalid/; s/rang/range/


> +            if (p->value->value < 1 || p->value->value > 255) {
> +                error_set(errp, QERR_INVALID_PARAMETER_VALUE,
> +                    "(de)compress_threads",
> +                    "is invalied, it should be in the rang of 1 to 255");

and again


> +++ b/monitor.c
> @@ -4544,6 +4544,24 @@ void migrate_set_capability_completion(ReadLineState *rs, int nb_args,
>      }
>  }
>  
> +void migrate_set_parameter_completion(ReadLineState *rs, int nb_args,
> +                                       const char *str)

Indentation is off.


> +++ b/qapi-schema.json
> @@ -540,6 +540,50 @@
>  ##
>  { 'command': 'query-migrate-capabilities', 'returns':   ['MigrationCapabilityStatus']}
>  
> +# @MigrationParameter
> +#
> +# Migration parameters enumeration
> +#
> +# @compress-level:Set the compression level to be used in live migration,

s/:/: /

> +#          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.
> +#
> +# Since: 2.3
> +##
> +{ 'enum': 'MigrationParameter',
> +  'data': ['compress-level', 'compress-threads', 'decompress-threads'] }
> +##
> +# @MigrationParameterStatus
> +#
> +# Migration parameter information
> +#
> +# @parameter: parameter enum
> +#
> +# @value: parameter value int
> +#
> +# Since: 2.3
> +##
> +{ 'type': 'MigrationParameterStatus',
> +  'data': { 'parameter' : 'MigrationParameter', 'value' : 'int' } }

Doesn't allow for non-integer parameters.  Not necessarily fatal for
input only (I think a flat union could be utilized with a
MigrationParameter as the discriminator to allow non-int values later on)...

> +##
> +# @migrate-set-parameters
> +#
> +# Set the following migration parameters (like compress-level)
> +#
> +# @parameters: json array of parameter modifications to make
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'migrate-set-parameters',
> +  'data': { 'parameters': ['MigrationParameterStatus'] } }

...but on output, we might confuse callers by outputting non-int values
unless we plan _from the start_ to support them.  That is, I think we want:

{ 'type': 'MigrationParameterBase',
  'data': { 'parameter': 'MigrationParameter' } }
{ 'type': 'MigrationParameterInt',
  'data': { 'value': 'int' } }
{ 'union': 'MigrationParameterStatus',
  'base': 'MigrationParameterBase',
  'discriminator': 'parameter',
  'data': { 'compress-level': 'MigrationParameterInt',
            'compress-threads': 'MigrationParameterInt',
            'decompress-threads': 'MigrationParameterInt' } }

to make it obvious to callers that they must be prepared to accept
non-int values down the road if we ever extend the union to add another
parameter with a different type.


>  SQMP
> +migrate-set-parameters
> +----------------------
> +
> +Set migration parameters
> +
> +- "compress-leve": multiple compression thread support

s/leve/level/

Missing two of the three defined parameters.

I hate write-only interfaces.  I'm hoping you add the query of
parameters in a later patch - but I'd prefer you squash it into one patch.

-- 
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] [v3 12/13] migration: Add command to set migration parameter

[Eric Blake]

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

On 01/23/2015 06:48 AM, Dr. David Alan Gilbert wrote:
> * Liang Li (liang.z.li@intel.com) wrote:
>> Add the qmp and hmp commands to tune the parameters used in live
>> migration.
> 
> If I understand correctly on the destination side we need to set the number of
> decompression threads very early on an incoming migration - I'm not
> clear how early that needs to be - especially if you're using fd: so it's
> not waiting for a connect ?
> 
> Eric: How would libvirt do that?

Libvirt does some handshaking before starting migration.  The source
would advertise that "I'm capable of threaded migration; and plan to use
X send and Y receive threads"; the destination would either reply that
threads are unsupported or set the receive thread count at that point,
then the source would know if it can enable threads and set the send
thread count, before letting migration start.  I don't see any problems
with the current design of starting things up.

-- 
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] [v3 13/13] migration: Add command to query migration parameter

[Eric Blake]

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

On 12/11/2014 06:29 PM, Liang Li wrote:
> Add the qmp and hmp commands to 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  |  2 ++
>  hmp.c            | 19 +++++++++++++++++++
>  hmp.h            |  1 +
>  migration.c      | 25 +++++++++++++++++++++++++
>  monitor.c        |  7 +++++++
>  qapi-schema.json | 10 ++++++++++
>  qmp-commands.hx  | 24 ++++++++++++++++++++++++
>  7 files changed, 88 insertions(+)
> 

> +    if (params) {
> +        monitor_printf(mon, "parameters: ");
> +        for (p = params; p; p = p->next) {
> +            monitor_printf(mon, "%s: %d ",
> +                           MigrationParameter_lookup[p->value->parameter],
> +                           (int)p->value->value);

Why are you casting to 'int'? I'm worried that this will fail if you
ever have a 64-bit parameter, or if we ever have a non-integral parameter.

> +++ b/qapi-schema.json
> @@ -584,6 +584,16 @@
>  { 'command': 'migrate-set-parameters',
>    'data': { 'parameters': ['MigrationParameterStatus'] } }
>  ##
> +# @query-migrate-parameters
> +#
> +# Returns information about the current migration parameters status
> +#
> +# Returns: @MigrationParametersStatus
> +#
> +# Since: 2.3
> +##
> +{ 'command': 'query-migrate-parameters', 'returns':   ['MigrationParameterStatus']}

Unusual spacing.  Line is longer than 80 columns.


> +query-migrate-parameters
> +------------------------
> +
> +Query current migration parameters
> +
> +- "parameters": migration parameters value
> +         - "compress-level" : compress level value (json-int)

Is it worth giving the example with all three defined parameters, rather
than truncating it to just one?

Again, see my comments on 12/13 about possibly squashing these together,
and/or using a union to make it obvious that we might support
non-integral parameters in the future.

-- 
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] [v3 12/13] migration: Add command to set migration parameter

[Dr. David Alan Gilbert]

* Eric Blake (eblake@redhat.com) wrote:
> On 01/23/2015 06:48 AM, Dr. David Alan Gilbert wrote:
> > * Liang Li (liang.z.li@intel.com) wrote:
> >> Add the qmp and hmp commands to tune the parameters used in live
> >> migration.
> > 
> > If I understand correctly on the destination side we need to set the number of
> > decompression threads very early on an incoming migration - I'm not
> > clear how early that needs to be - especially if you're using fd: so it's
> > not waiting for a connect ?
> > 
> > Eric: How would libvirt do that?
> 
> Libvirt does some handshaking before starting migration.  The source
> would advertise that "I'm capable of threaded migration; and plan to use
> X send and Y receive threads"; the destination would either reply that
> threads are unsupported or set the receive thread count at that point,
> then the source would know if it can enable threads and set the send
> thread count, before letting migration start.  I don't see any problems
> with the current design of starting things up.

Patch 3 calls migrate_decompress_threads_create from process_incoming_migration_co,
if you're using fd based incoming migration then I think that gets called
before libvirt would have a chance to connect to the monitor and
call the parameter setting to say how many threads it wants.
(A tcp incoming migration wouldn't have that problem because it
waits until the TCP accept before calling process_incoming_migration_co)

Dave

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


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

Re: [Qemu-devel] [v3 12/13] migration: Add command to set migration parameter

[Eric Blake]

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

On 01/23/2015 08:59 AM, Dr. David Alan Gilbert wrote:
> * Eric Blake (eblake@redhat.com) wrote:
>> On 01/23/2015 06:48 AM, Dr. David Alan Gilbert wrote:
>>> * Liang Li (liang.z.li@intel.com) wrote:
>>>> Add the qmp and hmp commands to tune the parameters used in live
>>>> migration.
>>>
>>> If I understand correctly on the destination side we need to set the number of
>>> decompression threads very early on an incoming migration - I'm not
>>> clear how early that needs to be - especially if you're using fd: so it's
>>> not waiting for a connect ?
>>>
>>> Eric: How would libvirt do that?
>>
>> Libvirt does some handshaking before starting migration.  The source
>> would advertise that "I'm capable of threaded migration; and plan to use
>> X send and Y receive threads"; the destination would either reply that
>> threads are unsupported or set the receive thread count at that point,
>> then the source would know if it can enable threads and set the send
>> thread count, before letting migration start.  I don't see any problems
>> with the current design of starting things up.
> 
> Patch 3 calls migrate_decompress_threads_create from process_incoming_migration_co,
> if you're using fd based incoming migration then I think that gets called
> before libvirt would have a chance to connect to the monitor and
> call the parameter setting to say how many threads it wants.
> (A tcp incoming migration wouldn't have that problem because it
> waits until the TCP accept before calling process_incoming_migration_co)

Then it probably needs to be exposed through a command-line parameter,
so that -incoming gains the ability to specify parameters as part of
starting up qemu.

-- 
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] [v3 02/13] migration: Add the framework of multi-thread compression

[Eric Blake]

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

On 12/11/2014 06:28 PM, Liang Li wrote:
> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---

> +void migrate_compress_threads_create(MigrationState *s)
> +{
> +    int i, thread_count;
> +
> +    if (!migrate_use_compression()) {
> +        return;
> +    }
> +    quit_thread = false;
> +    thread_count = migrate_compress_threads();
> +    s->compress_thread = g_malloc0(sizeof(QemuThread)
> +        * thread_count);

Theoretically unsafe (well, unsafe if thread_count were unbounded,
although it looks like you artificially cap it at 255 later in the
series); better would be:

s->compress_thread = g_new0(QemuThread, thread_count)

because that catches potential multiplication overflow.

> +    comp_param = g_malloc0(sizeof(compress_param) * thread_count);

Likewise.


>  
> +static int ram_save_compressed_page(QEMUFile *f, RAMBlock* block,

Spacing is off on the second '*'.

> +        ram_addr_t offset, bool last_stage)

Indentation is off.

-- 
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] [v3 03/13] migration: Add the framework of muti-thread decompression

[Eric Blake]

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

On 12/11/2014 06:28 PM, Liang Li wrote:

In addition to David's catches:

s/muti/multi/ in the subject line.

Commit message feels sparse - is the one subject line really all we need
to know about this framework?

> Signed-off-by: Liang Li <liang.z.li@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zhang@intel.com>
> ---
>  arch_init.c                   | 70 +++++++++++++++++++++++++++++++++++++++++++
>  include/migration/migration.h |  4 +++
>  migration.c                   | 15 ++++++++++
>  3 files changed, 89 insertions(+)
> 

>  
> +/* using compressBound() to calculate the buffer size needed to save the
> + * compressed data, to support the maximun TARGET_PAGE_SIZE bytes of
> + * data, we need more 15 bytes, use 16 to align the data.
> + */
> +#define COMPRESS_BUF_SIZE (TARGET_PAGE_SIZE + 16)
> +

>  static int ram_load(QEMUFile *f, void *opaque, int version_id)
>  {
>      int flags = 0, ret = 0;
>      static uint64_t seq_iter;
> +    int len = 0;
> +    uint8_t compbuf[COMPRESS_BUF_SIZE];
>  

Ouch - you stack-allocated more than a page of data.  This is in general
bad practice (and you should use the heap for any function that requires
more than 4k of data) because there are some architectures (cough:
windows) where exceeding the stack by more than a page risks silent
termination of the application rather than a graceful SIGSEGV (if you
can even call a stack overflow SIGSEGV graceful).  Especially true when
using helper threads, which typically have smaller stacks than the main
application.

>      seq_iter++;
>  
> @@ -1201,6 +1259,18 @@ 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("Illegal RAM offset " RAM_ADDR_FMT, addr);

s/Illegal/Invalid/ (the user isn't breaking a law, merely a constraint).

-- 
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] [PATCH v3 0/13] migration: Add a new feature to do live migration

[Li, Liang Z]

Thanks Dave & Eric for spending time to review my patches and giving the
valuable comments, I will refine my patches in the later version according 
to your suggestions.

> * Liang Li (liang.z.li@intel.com) wrote:
> > 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 v2->v3
> >
> >     -Splited the patch to 13 parts instead of 2
> >     -Rewrote the core code to do compression and decompression
> >     -Updated the document
> >     -Added a common command interface to set and query parameters
> >     -Added some comments
> 
> Hi,
>   Apologies for taking so long; generally this is a lot nicer.

> I'll reply to each patch individually, but two more general questions:
>    a) Shouldn't compressBound be used somewhere to get the worst case
> compressed size?

Yes, compressBound is better to deal with the worst case, I will make a change.

>    b) I think adding some trace points would be useful; it would make it easier
> to
>       spot when you were waiting for threads etc.
>

I will do that.

Liang

Re: [Qemu-devel] [v3 04/13] qemu-file: Add tow function will be used in migration

[Li, Liang Z]

> > +size_t migrate_qemu_add_compression_data(QEMUFile *f,
> > +        const uint8_t *p, size_t size, int level)
> 
> It's an odd name, QEMUFile is only used by migration anyway; maybe
> qemufile_add_compression_data ?
> 
> > +{
> > +    size_t blen = IO_BUF_SIZE - f->buf_index - sizeof(int);
> > +
> > +    if (compress2(f->buf + f->buf_index + sizeof(int), &blen, (Bytef *)p,
> > +            size, level) != Z_OK) {
> > +        error_report("Compress Failed!");
> > +        return 0;
> > +    }
> 
> What are the 'sizeof(int)'s for?  It's unusual because we normally keep the
> format of the stream the same even if one side was a 32bit qemu and the
> other 64bit.
> How do you know that there is enough space for the compress - i.e. what
> happens if f->buf_index is too high?

The compress2 will return failed if that happened. The spare space should be
checked before calling compress2, I will make a change.

> > +    qemu_put_be32(f, blen);
> > +    f->buf_index += blen;
> > +    return blen + sizeof(int);
> > +}
> > +
> > +int migrate_qemu_flush(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;
> > +}
> 
> Can you explain a bit more here how you're using the src file; I think you're
> using it as kind of a dummy buffer; but it needs to be documented
> somewhere.  Again I'm also not sure of the name of this function.
> 
Yes, it is for dummy buffer use, and I am not satisfied with the function name
too.

Re: [Qemu-devel] [v3 05/13] arch_init: alloc and free data struct in multi-thread compression

[Li, Liang Z]

> >  typedef struct compress_param compress_param;
> >
> > +enum {
> > +    DONE,
> > +    START,
> > +};
> > +
> 
> Do you really need any more than a 'bool busy' ?

Good ideal.

> >  struct decompress_param {
> >      /* To be done */
> >  };
> >  typedef struct decompress_param decompress_param;
> >
> >  static compress_param *comp_param;
> > +static QemuMutex *mutex;
> > +static QemuCond *cond;
> 
> Those need better names and a comment; If I'm reading it correctly, this
> cond is used to wake up the parent thread when one of the workers has
> finished it's task?

Yes, it is.
 

Re: [Qemu-devel] [v3 08/13] migration: Add the core code of multi-thread compresion

[Li, Liang Z]

> > -
> > +    /* 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.
> 
> Why?  Is this just because of the 'cont' flag used to avoid sending the block
> names again?
> 

Yes, it is. 

Re: [Qemu-devel] [v3 12/13] migration: Add command to set migration parameter

[Li, Liang Z]

 
> * Liang Li (liang.z.li@intel.com) wrote:
> > Add the qmp and hmp commands to tune the parameters used in live
> > migration.
> 
> If I understand correctly on the destination side we need to set the number
> of decompression threads very early on an incoming migration - I'm not clear
> how early that needs to be - especially if you're using fd: so it's not waiting
> for a connect ?

The decompression threads can be set after QEMU started (with -incomming
options) and before the TCP accept.

Re: [Qemu-devel] [v3 12/13] migration: Add command to set migration parameter

[Dr. David Alan Gilbert]

* Li, Liang Z (liang.z.li@intel.com) wrote:
>  
> > * Liang Li (liang.z.li@intel.com) wrote:
> > > Add the qmp and hmp commands to tune the parameters used in live
> > > migration.
> > 
> > If I understand correctly on the destination side we need to set the number
> > of decompression threads very early on an incoming migration - I'm not clear
> > how early that needs to be - especially if you're using fd: so it's not waiting
> > for a connect ?
> 
> The decompression threads can be set after QEMU started (with -incomming
> options) and before the TCP accept.

But that doesn't work for an fd: incoming migration.

Dave

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