* Peter Xu (pet...@redhat.com) wrote:
> To allow postcopy recovery, the ram fast load (preempt-only) dest QEMU thread
> needs similar handling on fault tolerance. When ram_load_postcopy() fails,
> instead of stopping the thread it halts with a semaphore, preparing to be
> kicked again when recovery is detected.
>
> A mutex is introduced to make sure there's no concurrent operation upon the
> socket. To make it simple, the fast ram load thread will take the mutex
> during
> its whole procedure, and only release it if it's paused. The fast-path socket
> will be properly released by the main loading thread safely when there's
> network failures during postcopy with that mutex held.
I *think* this is mostly OK; but I worry I don't understand all the
cases; e.g.
a) If the postcopy channel errors first
b) If the main channel errors first
Can you add some docs to walk through those and explain the locking ?
Dave
> Signed-off-by: Peter Xu <pet...@redhat.com>
> ---
> migration/migration.c | 17 +++++++++++++++--
> migration/migration.h | 3 +++
> migration/postcopy-ram.c | 24 ++++++++++++++++++++++--
> migration/savevm.c | 17 +++++++++++++++++
> migration/trace-events | 2 ++
> 5 files changed, 59 insertions(+), 4 deletions(-)
>
> diff --git a/migration/migration.c b/migration/migration.c
> index d20db04097..c68a281406 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -215,9 +215,11 @@ void migration_object_init(void)
> current_incoming->postcopy_remote_fds =
> g_array_new(FALSE, TRUE, sizeof(struct PostCopyFD));
> qemu_mutex_init(¤t_incoming->rp_mutex);
> + qemu_mutex_init(¤t_incoming->postcopy_prio_thread_mutex);
> qemu_event_init(¤t_incoming->main_thread_load_event, false);
> qemu_sem_init(¤t_incoming->postcopy_pause_sem_dst, 0);
> qemu_sem_init(¤t_incoming->postcopy_pause_sem_fault, 0);
> + qemu_sem_init(¤t_incoming->postcopy_pause_sem_fast_load, 0);
> qemu_mutex_init(¤t_incoming->page_request_mutex);
> current_incoming->page_requested = g_tree_new(page_request_addr_cmp);
>
> @@ -697,9 +699,9 @@ static bool postcopy_try_recover(void)
>
> /*
> * Here, we only wake up the main loading thread (while the
> - * fault thread will still be waiting), so that we can receive
> + * rest threads will still be waiting), so that we can receive
> * commands from source now, and answer it if needed. The
> - * fault thread will be woken up afterwards until we are sure
> + * rest threads will be woken up afterwards until we are sure
> * that source is ready to reply to page requests.
> */
> qemu_sem_post(&mis->postcopy_pause_sem_dst);
> @@ -3466,6 +3468,17 @@ static MigThrError postcopy_pause(MigrationState *s)
> qemu_file_shutdown(file);
> qemu_fclose(file);
>
> + /*
> + * Do the same to postcopy fast path socket too if there is. No
> + * locking needed because no racer as long as we do this before
> setting
> + * status to paused.
> + */
> + if (s->postcopy_qemufile_src) {
> +
> migration_ioc_unregister_yank_from_file(s->postcopy_qemufile_src);
Shouldn't this do a qemu_file_shutdown on here first?
> + qemu_fclose(s->postcopy_qemufile_src);
> + s->postcopy_qemufile_src = NULL;
> + }
> +
> migrate_set_state(&s->state, s->state,
> MIGRATION_STATUS_POSTCOPY_PAUSED);
>
> diff --git a/migration/migration.h b/migration/migration.h
> index b8aacfe3af..945088064a 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -118,6 +118,8 @@ struct MigrationIncomingState {
> /* Postcopy priority thread is used to receive postcopy requested pages
> */
> QemuThread postcopy_prio_thread;
> bool postcopy_prio_thread_created;
> + /* Used to sync with the prio thread */
> + QemuMutex postcopy_prio_thread_mutex;
> /*
> * An array of temp host huge pages to be used, one for each postcopy
> * channel.
> @@ -147,6 +149,7 @@ struct MigrationIncomingState {
> /* notify PAUSED postcopy incoming migrations to try to continue */
> QemuSemaphore postcopy_pause_sem_dst;
> QemuSemaphore postcopy_pause_sem_fault;
> + QemuSemaphore postcopy_pause_sem_fast_load;
>
> /* List of listening socket addresses */
> SocketAddressList *socket_address_list;
> diff --git a/migration/postcopy-ram.c b/migration/postcopy-ram.c
> index 30eddaacd1..b3d23804bc 100644
> --- a/migration/postcopy-ram.c
> +++ b/migration/postcopy-ram.c
> @@ -1575,6 +1575,15 @@ int postcopy_preempt_setup(MigrationState *s, Error
> **errp)
> return 0;
> }
>
> +static void postcopy_pause_ram_fast_load(MigrationIncomingState *mis)
> +{
> + trace_postcopy_pause_fast_load();
> + qemu_mutex_unlock(&mis->postcopy_prio_thread_mutex);
> + qemu_sem_wait(&mis->postcopy_pause_sem_fast_load);
> + qemu_mutex_lock(&mis->postcopy_prio_thread_mutex);
> + trace_postcopy_pause_fast_load_continued();
> +}
> +
> void *postcopy_preempt_thread(void *opaque)
> {
> MigrationIncomingState *mis = opaque;
> @@ -1587,11 +1596,22 @@ void *postcopy_preempt_thread(void *opaque)
> qemu_sem_post(&mis->thread_sync_sem);
>
> /* Sending RAM_SAVE_FLAG_EOS to terminate this thread */
> - ret = ram_load_postcopy(mis->postcopy_qemufile_dst,
> RAM_CHANNEL_POSTCOPY);
> + qemu_mutex_lock(&mis->postcopy_prio_thread_mutex);
> + while (1) {
> + ret = ram_load_postcopy(mis->postcopy_qemufile_dst,
> RAM_CHANNEL_POSTCOPY);
> + /* If error happened, go into recovery routine */
> + if (ret) {
> + postcopy_pause_ram_fast_load(mis);
> + } else {
> + /* We're done */
> + break;
> + }
> + }
> + qemu_mutex_unlock(&mis->postcopy_prio_thread_mutex);
>
> rcu_unregister_thread();
>
> trace_postcopy_preempt_thread_exit();
>
> - return ret == 0 ? NULL : (void *)-1;
> + return NULL;
> }
> diff --git a/migration/savevm.c b/migration/savevm.c
> index 254aa78234..2d32340d28 100644
> --- a/migration/savevm.c
> +++ b/migration/savevm.c
> @@ -2152,6 +2152,13 @@ static int
> loadvm_postcopy_handle_resume(MigrationIncomingState *mis)
> */
> qemu_sem_post(&mis->postcopy_pause_sem_fault);
>
> + if (migrate_postcopy_preempt()) {
> + /* The channel should already be setup again; make sure of it */
> + assert(mis->postcopy_qemufile_dst);
> + /* Kick the fast ram load thread too */
> + qemu_sem_post(&mis->postcopy_pause_sem_fast_load);
> + }
> +
> return 0;
> }
>
> @@ -2607,6 +2614,16 @@ static bool
> postcopy_pause_incoming(MigrationIncomingState *mis)
> mis->to_src_file = NULL;
> qemu_mutex_unlock(&mis->rp_mutex);
>
> + if (mis->postcopy_qemufile_dst) {
> + qemu_file_shutdown(mis->postcopy_qemufile_dst);
> + /* Take the mutex to make sure the fast ram load thread halted */
> + qemu_mutex_lock(&mis->postcopy_prio_thread_mutex);
> + migration_ioc_unregister_yank_from_file(mis->postcopy_qemufile_dst);
> + qemu_fclose(mis->postcopy_qemufile_dst);
> + mis->postcopy_qemufile_dst = NULL;
> + qemu_mutex_unlock(&mis->postcopy_prio_thread_mutex);
> + }
> +
> migrate_set_state(&mis->state, MIGRATION_STATUS_POSTCOPY_ACTIVE,
> MIGRATION_STATUS_POSTCOPY_PAUSED);
>
> diff --git a/migration/trace-events b/migration/trace-events
> index b1155d9da0..dfe36a3340 100644
> --- a/migration/trace-events
> +++ b/migration/trace-events
> @@ -270,6 +270,8 @@ mark_postcopy_blocktime_begin(uint64_t addr, void *dd,
> uint32_t time, int cpu, i
> mark_postcopy_blocktime_end(uint64_t addr, void *dd, uint32_t time, int
> affected_cpu) "addr: 0x%" PRIx64 ", dd: %p, time: %u, affected_cpu: %d"
> postcopy_pause_fault_thread(void) ""
> postcopy_pause_fault_thread_continued(void) ""
> +postcopy_pause_fast_load(void) ""
> +postcopy_pause_fast_load_continued(void) ""
> postcopy_ram_fault_thread_entry(void) ""
> postcopy_ram_fault_thread_exit(void) ""
> postcopy_ram_fault_thread_fds_core(int baseufd, int quitfd) "ufd: %d quitfd:
> %d"
> --
> 2.32.0
>
--
Dr. David Alan Gilbert / dgilb...@redhat.com / Manchester, UK
