1 files changed, 122 insertions, 0 deletions

diff --git a/migration/migration.c b/migration/migration.c
index bc42490..732d229 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -33,6 +33,14 @@
 #define BUFFER_DELAY     100
 #define XFER_LIMIT_RATIO (1000 / BUFFER_DELAY)
 
+/* Default compression thread count */
+#define DEFAULT_MIGRATE_COMPRESS_THREAD_COUNT 8
+/* Default decompression thread count, usually decompression is at
+ * least 4 times as fast as compression.*/
+#define DEFAULT_MIGRATE_DECOMPRESS_THREAD_COUNT 2
+/*0: means nocompress, 1: best speed, ... 9: best compress ratio */
+#define DEFAULT_MIGRATE_COMPRESS_LEVEL 1
+
 /* Migration XBZRLE default cache size */
 #define DEFAULT_MIGRATE_CACHE_SIZE (64 * 1024 * 1024)
 
@@ -52,6 +60,12 @@ MigrationState *migrate_get_current(void)
         .bandwidth_limit = MAX_THROTTLE,
         .xbzrle_cache_size = DEFAULT_MIGRATE_CACHE_SIZE,
         .mbps = -1,
+        .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;
@@ -106,6 +120,7 @@ static void process_incoming_migration_co(void *opaque)
     free_xbzrle_decoded_buf();
     if (ret < 0) {
         error_report("load of migration failed: %s", strerror(-ret));
+        migrate_decompress_threads_join();
         exit(EXIT_FAILURE);
     }
     qemu_announce_self();
@@ -114,6 +129,7 @@ static void process_incoming_migration_co(void *opaque)
     bdrv_invalidate_cache_all(&local_err);
     if (local_err) {
         error_report_err(local_err);
+        migrate_decompress_threads_join();
         exit(EXIT_FAILURE);
     }
 
@@ -122,6 +138,7 @@ static void process_incoming_migration_co(void *opaque)
     } else {
         runstate_set(RUN_STATE_PAUSED);
     }
+    migrate_decompress_threads_join();
 }
 
 void process_incoming_migration(QEMUFile *f)
@@ -130,6 +147,7 @@ void process_incoming_migration(QEMUFile *f)
     int fd = qemu_get_fd(f);
 
     assert(fd != -1);
+    migrate_decompress_threads_create();
     qemu_set_nonblock(fd);
     qemu_coroutine_enter(co, f);
 }
@@ -170,6 +188,21 @@ MigrationCapabilityStatusList *qmp_query_migrate_capabilities(Error **errp)
     return head;
 }
 
+MigrationParameters *qmp_query_migrate_parameters(Error **errp)
+{
+    MigrationParameters *params;
+    MigrationState *s = migrate_get_current();
+
+    params = g_malloc0(sizeof(*params));
+    params->compress_level = s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
+    params->compress_threads =
+            s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
+    params->decompress_threads =
+            s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
+
+    return params;
+}
+
 static void get_xbzrle_cache_stats(MigrationInfo *info)
 {
     if (migrate_use_xbzrle()) {
@@ -283,6 +316,47 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
     }
 }
 
+void qmp_migrate_set_parameters(bool has_compress_level,
+                                int64_t compress_level,
+                                bool has_compress_threads,
+                                int64_t compress_threads,
+                                bool has_decompress_threads,
+                                int64_t decompress_threads, Error **errp)
+{
+    MigrationState *s = migrate_get_current();
+
+    if (has_compress_level && (compress_level < 0 || compress_level > 9)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE, "compress_level",
+                  "is invalid, it should be in the range of 0 to 9");
+        return;
+    }
+    if (has_compress_threads &&
+            (compress_threads < 1 || compress_threads > 255)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                  "compress_threads",
+                  "is invalid, it should be in the range of 1 to 255");
+        return;
+    }
+    if (has_decompress_threads &&
+            (decompress_threads < 1 || decompress_threads > 255)) {
+        error_set(errp, QERR_INVALID_PARAMETER_VALUE,
+                  "decompress_threads",
+                  "is invalid, it should be in the range of 1 to 255");
+        return;
+    }
+
+    if (has_compress_level) {
+        s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
+    }
+    if (has_compress_threads) {
+        s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] = compress_threads;
+    }
+    if (has_decompress_threads) {
+        s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+                                                    decompress_threads;
+    }
+}
+
 /* shared migration helpers */
 
 static void migrate_set_state(MigrationState *s, int old_state, int new_state)
@@ -305,6 +379,7 @@ static void migrate_fd_cleanup(void *opaque)
         qemu_thread_join(&s->thread);
         qemu_mutex_lock_iothread();
 
+        migrate_compress_threads_join();
         qemu_fclose(s->file);
         s->file = NULL;
     }
@@ -390,6 +465,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->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));
@@ -400,6 +480,11 @@ static MigrationState *migrate_init(const MigrationParams *params)
            sizeof(enabled_capabilities));
     s->xbzrle_cache_size = xbzrle_cache_size;
 
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL] = compress_level;
+    s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS] =
+               compress_thread_count;
+    s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS] =
+               decompress_thread_count;
     s->bandwidth_limit = bandwidth_limit;
     s->state = MIGRATION_STATUS_SETUP;
     trace_migrate_set_state(MIGRATION_STATUS_SETUP);
@@ -587,6 +672,42 @@ bool migrate_zero_blocks(void)
     return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS];
 }
 
+bool migrate_use_compression(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->enabled_capabilities[MIGRATION_CAPABILITY_COMPRESS];
+}
+
+int migrate_compress_level(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->parameters[MIGRATION_PARAMETER_COMPRESS_LEVEL];
+}
+
+int migrate_compress_threads(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->parameters[MIGRATION_PARAMETER_COMPRESS_THREADS];
+}
+
+int migrate_decompress_threads(void)
+{
+    MigrationState *s;
+
+    s = migrate_get_current();
+
+    return s->parameters[MIGRATION_PARAMETER_DECOMPRESS_THREADS];
+}
+
 int migrate_use_xbzrle(void)
 {
     MigrationState *s;
@@ -730,6 +851,7 @@ void migrate_fd_connect(MigrationState *s)
     /* Notify before starting migration thread */
     notifier_list_notify(&migration_state_notifiers, s);
 
+    migrate_compress_threads_create();
     qemu_thread_create(&s->thread, "migration", migration_thread, s,
                        QEMU_THREAD_JOINABLE);
 }