| Message ID | 1423623986-590-3-git-send-email-liang.z.li@intel.com |
|---|---|
| State | New |
| Headers | show |
Liang Li <liang.z.li@intel.com> wrote: > Add the code to create and destroy the multiple threads those will > be used to do data compression. Left some functions empty to keep > clearness, and the code will be added later. > > Signed-off-by: Liang Li <liang.z.li@intel.com> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com> Reviewed-by: Juan Quintela <quintela@redhat.com> > +int migrate_compress_threads(void) > +{ > + MigrationState *s; > + > + s = migrate_get_current(); > + > + return s->compress_thread_count; > +} > + As far as I can see, all users have already s in place. No that I disagree with an acessor function.
Liang Li <liang.z.li@intel.com> wrote: > Add the code to create and destroy the multiple threads those will > be used to do data compression. Left some functions empty to keep > clearness, and the code will be added later. > > Signed-off-by: Liang Li <liang.z.li@intel.com> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com> /me goes back > +struct CompressParam { > + /* To be done */ > +}; > +typedef struct CompressParam CompressParam; > + > +static CompressParam *comp_param; > +static bool quit_comp_thread; > + > +static void *do_data_compress(void *opaque) > +{ > + while (!quit_comp_thread) { Using this variable without any protection, read from all threads. > + > + /* To be done */ > + > + } > + > + return NULL; > +} > + > +static inline void terminate_compression_threads(void) > +{ > + quit_comp_thread = true; written from the main migration thread. Shouldn't we use a mutex or at least atomic accessors? PD. I haven't yet finishing reading the series, in case this is fixed later.
Liang Li <liang.z.li@intel.com> wrote: > Add the code to create and destroy the multiple threads those will > be used to do data compression. Left some functions empty to keep > clearness, and the code will be added later. > > Signed-off-by: Liang Li <liang.z.li@intel.com> > Signed-off-by: Yang Zhang <yang.z.zhang@intel.com> > Reviewed-by: Dr.David Alan Gilbert <dgilbert@redhat.com> And here I am again. Reviewing patch 8, I found that we need to fix some things here. > +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block, > + ram_addr_t offset, bool last_stage) > +{ > + int bytes_sent = -1; > + > + /* To be done*/ > + > + return bytes_sent; > +} We have three return values, here, that are not the same that for normal pages 0: this is the 1st page for a particular thread, nothing to sent yet n > 0: we are sending the previous compresed page for the choosen thread Notice that the only way that ram_save_page() can return 0 is for xbzrle when a page has modified but it has exactly the same value that before. (it can have been modified twice, +1, -1 or whatever) Notice that ram_save_page() can only return 0 (duplicate page) or > 0 (real size written) > + > /* > * ram_find_and_save_block: Finds a page to send and sends it to f > * > @@ -679,7 +751,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); > + } I need more context, this is the corrent code } else { bytes_sent = ram_save_page(f, block, offset, last_stage); /* if page is unmodified, continue to the next */ if (bytes_sent > 0) { last_sent_block = block; break; } } And we should change to: } else if (migrate_use_compression()) { bytes_sent = ram_save_compressed_page(f, block, offset, last_stage); last_sent_block = block; break; } else { bytes_sent = ram_save_page(f, block, offset, last_stage); /* if page is unmodified, continue to the next */ if (bytes_sent > 0) { last_sent_block = block; break; } } This would mean that we don't need to arrange for the zero byte return on qemu.
> Reviewing patch 8, I found that we need to fix some things here. > > > +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block, > > + ram_addr_t offset, bool > > +last_stage) { > > + int bytes_sent = -1; > > + > > + /* To be done*/ > > + > > + return bytes_sent; > > +} > > We have three return values, here, that are not the same that for normal > pages > > 0: this is the 1st page for a particular thread, nothing to sent yet n > 0: we > are sending the previous compresed page for the choosen > thread > > Notice that the only way that ram_save_page() can return 0 is for xbzrle > when a page has modified but it has exactly the same value that before. > > (it can have been modified twice, +1, -1 or whatever) > > Notice that ram_save_page() can only return 0 (duplicate page) or > 0 (real > size written) > > > + > > /* > > * ram_find_and_save_block: Finds a page to send and sends it to f > > * > > @@ -679,7 +751,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); > > + } > > > I need more context, this is the corrent code > > } else { > bytes_sent = ram_save_page(f, block, offset, last_stage); > > /* if page is unmodified, continue to the next */ > if (bytes_sent > 0) { > last_sent_block = block; > break; > } > } > > And we should change to: > > } else if (migrate_use_compression()) { > bytes_sent = ram_save_compressed_page(f, block, offset, > last_stage); > last_sent_block = block; > break; What happened if ram_save_compressed_page() return 0 ? following the flush_compressed_data() will be call, The code call still work, but the efficiency is poor. Every time the main thread find there is no free compression thread, it has to wait all compression threads finish their job before it can start the next round. The effective way is to start compression once there is any free compression thread. Liang > } else { > bytes_sent = ram_save_page(f, block, offset, last_stage); > > /* if page is unmodified, continue to the next */ > if (bytes_sent > 0) { > last_sent_block = block; > break; > } > } > > This would mean that we don't need to arrange for the zero byte return on > qemu. >
"Li, Liang Z" <liang.z.li@intel.com> wrote: >> Reviewing patch 8, I found that we need to fix some things here. >> >> > +static int ram_save_compressed_page(QEMUFile *f, RAMBlock *block, >> > + ram_addr_t offset, bool >> > +last_stage) { >> > + int bytes_sent = -1; >> > + >> > + /* To be done*/ >> > + >> > + return bytes_sent; >> > +} >> >> We have three return values, here, that are not the same that for normal >> pages >> >> 0: this is the 1st page for a particular thread, nothing to sent yet n > 0: we >> are sending the previous compresed page for the choosen >> thread >> >> Notice that the only way that ram_save_page() can return 0 is for xbzrle >> when a page has modified but it has exactly the same value that before. >> >> (it can have been modified twice, +1, -1 or whatever) >> >> Notice that ram_save_page() can only return 0 (duplicate page) or > 0 (real >> size written) >> >> > + >> > /* >> > * ram_find_and_save_block: Finds a page to send and sends it to f >> > * >> > @@ -679,7 +751,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); >> > + } >> >> >> I need more context, this is the corrent code >> >> } else { >> bytes_sent = ram_save_page(f, block, offset, last_stage); >> >> /* if page is unmodified, continue to the next */ >> if (bytes_sent > 0) { >> last_sent_block = block; >> break; >> } >> } >> >> And we should change to: >> >> } else if (migrate_use_compression()) { >> bytes_sent = ram_save_compressed_page(f, block, offset, >> last_stage); >> last_sent_block = block; >> break; > > > What happened if ram_save_compressed_page() return 0 ? following the > flush_compressed_data() will be call, This happens to me to send suggestions instead of proper code. You are right. I fixed one of the callers, but not the "upstream" caller. > The code call still work, but the efficiency is poor. Every time the > main thread find there is no free compression > thread, it has to wait all compression threads finish their job before > it can start the next round. > The effective way is to start compression once there is any free > compression thread. Will send to the list a suggestion to improve from here, right? Sorry for the noise, Juan. > Liang > >> } else { >> bytes_sent = ram_save_page(f, block, offset, last_stage); >> >> /* if page is unmodified, continue to the next */ >> if (bytes_sent > 0) { >> last_sent_block = block; >> break; >> } >> } >> >> This would mean that we don't need to arrange for the zero byte return on >> qemu. >>
diff --git a/arch_init.c b/arch_init.c index 89c8fa4..709036c 100644 --- a/arch_init.c +++ b/arch_init.c @@ -332,6 +332,68 @@ static uint64_t migration_dirty_pages; static uint32_t last_version; static bool ram_bulk_stage; +struct CompressParam { + /* To be done */ +}; +typedef struct CompressParam CompressParam; + +static CompressParam *comp_param; +static bool quit_comp_thread; + +static void *do_data_compress(void *opaque) +{ + while (!quit_comp_thread) { + + /* To be done */ + + } + + return NULL; +} + +static inline void terminate_compression_threads(void) +{ + quit_comp_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_comp_thread = false; + thread_count = migrate_compress_threads(); + s->compress_thread = g_new0(QemuThread, thread_count); + comp_param = g_new0(CompressParam, 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. @@ -645,6 +707,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 = -1; + + /* To be done*/ + + return bytes_sent; +} + /* * ram_find_and_save_block: Finds a page to send and sends it to f * @@ -679,7 +751,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 f37348b..228badb 100644 --- a/include/migration/migration.h +++ b/include/migration/migration.h @@ -50,6 +50,9 @@ struct MigrationState QemuThread thread; QEMUBH *cleanup_bh; QEMUFile *file; + QemuThread *compress_thread; + int compress_thread_count; + int compress_level; int state; MigrationParams params; @@ -108,6 +111,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); @@ -157,6 +162,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/migration.c b/migration/migration.c index b3adbc6..309443e 100644 --- a/migration/migration.c +++ b/migration/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 ¤t_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; } @@ -385,6 +393,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)); @@ -395,6 +405,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); @@ -567,6 +579,30 @@ bool migrate_zero_blocks(void) return s->enabled_capabilities[MIGRATION_CAPABILITY_ZERO_BLOCKS]; } +bool migrate_use_compression(void) +{ + /* Disable compression before the patch series 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; @@ -707,6 +743,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); }