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

[Liang Li]

Signed-off-by: Liang Li <address@hidden>
Signed-off-by: Yang Zhang <address@hidden>
---
 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