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

[Dr. David Alan Gilbert]

* Liang Li (address@hidden) wrote:
> Add the code to create and destroy the multiple threads those will be
> used to do data decompression. Left some functions empty just to keep
> clearness, and the code will be added later.

Reviewed-by: Dr. David Alan Gilbert <address@hidden>

> 
> Signed-off-by: Liang Li <address@hidden>
> Signed-off-by: Yang Zhang <address@hidden>
> ---
>  arch_init.c                   | 75 
> +++++++++++++++++++++++++++++++++++++++++++
>  include/migration/migration.h |  4 +++
>  migration/migration.c         | 16 +++++++++
>  3 files changed, 95 insertions(+)
> 
> diff --git a/arch_init.c b/arch_init.c
> index 1831f1a..ed34eb3 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>
> @@ -126,6 +127,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;
> @@ -337,8 +339,16 @@ struct CompressParam {
>  };
>  typedef struct CompressParam CompressParam;
>  
> +struct DecompressParam {
> +    /* To be done */
> +};
> +typedef struct DecompressParam DecompressParam;
> +
>  static CompressParam *comp_param;
>  static bool quit_thread;
> +static DecompressParam *decomp_param;
> +static QemuThread *decompress_threads;
> +static uint8_t *compressed_data_buf;
>  
>  static void *do_data_compress(void *opaque)
>  {
> @@ -1128,10 +1138,58 @@ 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_new0(QemuThread, count);
> +    decomp_param = g_new0(DecompressParam, count);
> +    compressed_data_buf = g_malloc0(compressBound(TARGET_PAGE_SIZE));
> +    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);
> +    g_free(compressed_data_buf);
> +    decompress_threads = NULL;
> +    decomp_param = NULL;
> +    compressed_data_buf = 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;
>  
>      seq_iter++;
>  
> @@ -1208,6 +1266,23 @@ 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("Invalid RAM offset " RAM_ADDR_FMT, addr);
> +                ret = -EINVAL;
> +                break;
> +            }
> +
> +            len = qemu_get_be32(f);
> +            if (len < 0 || len > compressBound(TARGET_PAGE_SIZE)) {
> +                error_report("Invalid compressed data length: %d", len);
> +                ret = -EINVAL;
> +                break;
> +            }
> +            qemu_get_buffer(f, compressed_data_buf, len);
> +            decompress_data_with_multi_threads(compressed_data_buf, 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/migration.c b/migration/migration.c
> index 309443e..a6f6e02 100644
> --- a/migration/migration.c
> +++ b/migration/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,12 +125,15 @@ static void process_incoming_migration_co(void *opaque)
>      } else {
>          runstate_set(RUN_STATE_PAUSED);
>      }
> +    migrate_decompress_threads_join();
>  }
>  
>  void process_incoming_migration(QEMUFile *f)
>  {
>      Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
>      int fd = qemu_get_fd(f);
> +    int thread_count = migrate_decompress_threads();
> +    migrate_decompress_threads_create(thread_count);
>  
>      assert(fd != -1);
>      qemu_set_nonblock(fd);
> @@ -395,6 +400,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));
> @@ -407,6 +413,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);
> @@ -603,6 +610,15 @@ 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)
>  {
>      MigrationState *s;
> -- 
> 1.9.1
> 
--
Dr. David Alan Gilbert / address@hidden / Manchester, UK