[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [v3 03/13] migration: Add the framework of muti-thread deco
From: |
Liang Li |
Subject: |
[Qemu-devel] [v3 03/13] migration: Add the framework of muti-thread decompression |
Date: |
Fri, 12 Dec 2014 09:28:56 +0800 |
Signed-off-by: Liang Li <address@hidden>
Signed-off-by: Yang Zhang <address@hidden>
---
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] [PATCH v3 0/13] migration: Add a new feature to do live migration, Liang Li, 2014/12/11
- [Qemu-devel] [v3 01/13] docs: Add a doc about multiple thread compression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 04/13] qemu-file: Add tow function will be used in migration, Liang Li, 2014/12/11
- [Qemu-devel] [v3 05/13] arch_init: alloc and free data struct in multi-thread compression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 02/13] migration: Add the framework of multi-thread compression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 06/13] arch_init: Add data struct used by decompression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 03/13] migration: Add the framework of muti-thread decompression,
Liang Li <=
- [Qemu-devel] [v3 07/13] migraion: Rewrite the function ram_save_page(), Liang Li, 2014/12/11
- [Qemu-devel] [v3 08/13] migration: Add the core code of multi-thread compresion, Liang Li, 2014/12/11
- [Qemu-devel] [v3 09/13] migration: Make compression co-work with xbzrle, Liang Li, 2014/12/11
- [Qemu-devel] [v3 10/13] migration: Add the core code of multi-thread decompression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 11/13] migration: Add interface to control compression, Liang Li, 2014/12/11
- [Qemu-devel] [v3 12/13] migration: Add command to set migration parameter, Liang Li, 2014/12/11
- [Qemu-devel] [v3 13/13] migration: Add command to query migration parameter, Liang Li, 2014/12/11
- Re: [Qemu-devel] [PATCH v3 0/13] migration: Add a new feature to do live migration, Li, Liang Z, 2014/12/24