Re: [PATCH 19/20] migration: Postcopy recover with preempt enabled

[Dr. David Alan Gilbert]

* Peter Xu (peterx@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 <peterx@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(&current_incoming->rp_mutex);
> +    qemu_mutex_init(&current_incoming->postcopy_prio_thread_mutex);
>      qemu_event_init(&current_incoming->main_thread_load_event, false);
>      qemu_sem_init(&current_incoming->postcopy_pause_sem_dst, 0);
>      qemu_sem_init(&current_incoming->postcopy_pause_sem_fault, 0);
> +    qemu_sem_init(&current_incoming->postcopy_pause_sem_fast_load, 0);
>      qemu_mutex_init(&current_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 / dgilbert@redhat.com / Manchester, UK