* Liang Li (liang.z...@intel.com) wrote:
> Signed-off-by: Liang Li <liang.z...@intel.com>
> Signed-off-by: Yang Zhang <yang.z.zh...@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 / dgilb...@redhat.com / Manchester, UK
