Re: [PATCH 3/8] migration: Add precopy initial data loaded ACK functionality

[Avihai Horon <avihaih@nvidia.com>]

On 03/05/2023 1:56, Peter Xu wrote:
> External email: Use caution opening links or attachments
>
>
> On Mon, May 01, 2023 at 05:01:36PM +0300, Avihai Horon wrote:
>> Add the core functionality of precopy initial data, which allows the
>> destination to ACK that initial data has been loaded and the source to
>> wait for this ACK before completing the migration.
>>
>> A new return path command MIG_RP_MSG_INITIAL_DATA_LOADED_ACK is added.
>> It is sent by the destination after precopy initial data is loaded to
>> ACK to the source that precopy initial data has been loaded.
>>
>> In addition, two new SaveVMHandlers handlers are added:
>> 1. is_initial_data_active which indicates whether precopy initial data
>>     is used for this migration user (i.e., SaveStateEntry).
>> 2. initial_data_loaded which indicates whether precopy initial data has
>>     been loaded by this migration user.
>>
>> Signed-off-by: Avihai Horon <avihaih@nvidia.com>
>> ---
>>   include/migration/register.h |  7 ++++++
>>   migration/migration.h        | 12 +++++++++++
>>   migration/migration.c        | 41 ++++++++++++++++++++++++++++++++++--
>>   migration/savevm.c           | 39 ++++++++++++++++++++++++++++++++++
>>   migration/trace-events       |  2 ++
>>   5 files changed, 99 insertions(+), 2 deletions(-)
>>
>> diff --git a/include/migration/register.h b/include/migration/register.h
>> index 0a73f3883e..297bbe9f73 100644
>> --- a/include/migration/register.h
>> +++ b/include/migration/register.h
>> @@ -77,6 +77,13 @@ typedef struct SaveVMHandlers {
>>        * true if it's supported or false otherwise. Called both in src and dest.
>>        */
>>       bool (*initial_data_advise)(void *opaque);
>> +    /*
>> +     * Checks if precopy initial data is active. If it's inactive,
>> +     * initial_data_loaded check is skipped.
>> +     */
>> +    bool (*is_initial_data_active)(void *opaque);
>> +    /* Checks if precopy initial data has been loaded in dest */
>> +    bool (*initial_data_loaded)(void *opaque);
>>   } SaveVMHandlers;
>>
>>   int register_savevm_live(const char *idstr,
>> diff --git a/migration/migration.h b/migration/migration.h
>> index 4f615e9dbc..d865c23d87 100644
>> --- a/migration/migration.h
>> +++ b/migration/migration.h
>> @@ -207,6 +207,11 @@ struct MigrationIncomingState {
>>
>>       /* Indicates whether precopy initial data was enabled and should be used */
>>       bool initial_data_enabled;
>> +    /*
>> +     * Indicates whether an ACK that precopy initial data was loaded has been
>> +     * sent to source.
>> +     */
>> +    bool initial_data_loaded_ack_sent;
>>   };
>>
>>   MigrationIncomingState *migration_incoming_get_current(void);
>> @@ -435,6 +440,12 @@ struct MigrationState {
>>
>>       /* QEMU_VM_VMDESCRIPTION content filled for all non-iterable devices. */
>>       JSONWriter *vmdesc;
>> +
>> +    /*
>> +     * Indicates whether an ACK that precopy initial data was loaded in
>> +     * destination has been received.
>> +     */
>> +    bool initial_data_loaded_acked;
>>   };
>>
>>   void migrate_set_state(int *state, int old_state, int new_state);
>> @@ -475,6 +486,7 @@ int migrate_send_rp_message_req_pages(MigrationIncomingState *mis,
>>   void migrate_send_rp_recv_bitmap(MigrationIncomingState *mis,
>>                                    char *block_name);
>>   void migrate_send_rp_resume_ack(MigrationIncomingState *mis, uint32_t value);
>> +int migrate_send_rp_initial_data_loaded_ack(MigrationIncomingState *mis);
>>
>>   void dirty_bitmap_mig_before_vm_start(void);
>>   void dirty_bitmap_mig_cancel_outgoing(void);
>> diff --git a/migration/migration.c b/migration/migration.c
>> index 68cdf5b184..304cab2fa1 100644
>> --- a/migration/migration.c
>> +++ b/migration/migration.c
>> @@ -77,6 +77,11 @@ enum mig_rp_message_type {
>>       MIG_RP_MSG_RECV_BITMAP,  /* send recved_bitmap back to source */
>>       MIG_RP_MSG_RESUME_ACK,   /* tell source that we are ready to resume */
>>
>> +    MIG_RP_MSG_INITIAL_DATA_LOADED_ACK, /*
>> +                                         * Tell source precopy initial data is
>> +                                         * loaded.
>> +                                         */
>> +
>>       MIG_RP_MSG_MAX
>>   };
>>
>> @@ -756,6 +761,12 @@ bool migration_has_all_channels(void)
>>       return true;
>>   }
>>
>> +int migrate_send_rp_initial_data_loaded_ack(MigrationIncomingState *mis)
>> +{
>> +    return migrate_send_rp_message(mis, MIG_RP_MSG_INITIAL_DATA_LOADED_ACK, 0,
>> +                                   NULL);
>> +}
>> +
>>   /*
>>    * Send a 'SHUT' message on the return channel with the given value
>>    * to indicate that we've finished with the RP.  Non-0 value indicates
>> @@ -1401,6 +1412,8 @@ void migrate_init(MigrationState *s)
>>       s->vm_was_running = false;
>>       s->iteration_initial_bytes = 0;
>>       s->threshold_size = 0;
>> +
>> +    s->initial_data_loaded_acked = false;
>>   }
>>
>>   int migrate_add_blocker_internal(Error *reason, Error **errp)
>> @@ -1717,6 +1730,9 @@ static struct rp_cmd_args {
>>       [MIG_RP_MSG_REQ_PAGES_ID]   = { .len = -1, .name = "REQ_PAGES_ID" },
>>       [MIG_RP_MSG_RECV_BITMAP]    = { .len = -1, .name = "RECV_BITMAP" },
>>       [MIG_RP_MSG_RESUME_ACK]     = { .len =  4, .name = "RESUME_ACK" },
>> +    [MIG_RP_MSG_INITIAL_DATA_LOADED_ACK] = { .len = 0,
>> +                                             .name =
>> +                                                 "INITIAL_DATA_LOADED_ACK" },
>>       [MIG_RP_MSG_MAX]            = { .len = -1, .name = "MAX" },
>>   };
>>
>> @@ -1955,6 +1971,11 @@ retry:
>>               }
>>               break;
>>
>> +        case MIG_RP_MSG_INITIAL_DATA_LOADED_ACK:
>> +            ms->initial_data_loaded_acked = true;
>> +            trace_source_return_path_thread_initial_data_loaded_ack();
>> +            break;
>> +
>>           default:
>>               break;
>>           }
>> @@ -2704,6 +2725,20 @@ static void migration_update_counters(MigrationState *s,
>>                                 bandwidth, s->threshold_size);
>>   }
>>
>> +static bool initial_data_loaded_acked(MigrationState *s)
>> +{
>> +    if (!migrate_precopy_initial_data()) {
>> +        return true;
>> +    }
>> +
>> +    /* No reason to wait for precopy initial data loaded ACK if VM is stopped */
>> +    if (!runstate_is_running()) {
>> +        return true;
>> +    }
>> +
>> +    return s->initial_data_loaded_acked;
>> +}
>> +
>>   /* Migration thread iteration status */
>>   typedef enum {
>>       MIG_ITERATE_RESUME,         /* Resume current iteration */
>> @@ -2719,6 +2754,7 @@ static MigIterateState migration_iteration_run(MigrationState *s)
>>   {
>>       uint64_t must_precopy, can_postcopy;
>>       bool in_postcopy = s->state == MIGRATION_STATUS_POSTCOPY_ACTIVE;
>> +    bool initial_data_loaded = initial_data_loaded_acked(s);
>>
>>       qemu_savevm_state_pending_estimate(&must_precopy, &can_postcopy);
>>       uint64_t pending_size = must_precopy + can_postcopy;
>> @@ -2731,7 +2767,8 @@ static MigIterateState migration_iteration_run(MigrationState *s)
>>           trace_migrate_pending_exact(pending_size, must_precopy, can_postcopy);
>>       }
>>
>> -    if (!pending_size || pending_size < s->threshold_size) {
>> +    if ((!pending_size || pending_size < s->threshold_size) &&
>> +        initial_data_loaded) {
>>           trace_migration_thread_low_pending(pending_size);
>>           migration_completion(s);
>>           return MIG_ITERATE_BREAK;
>> @@ -2739,7 +2776,7 @@ static MigIterateState migration_iteration_run(MigrationState *s)
>>
>>       /* Still a significant amount to transfer */
>>       if (!in_postcopy && must_precopy <= s->threshold_size &&
>> -        qatomic_read(&s->start_postcopy)) {
>> +        initial_data_loaded && qatomic_read(&s->start_postcopy)) {
>>           if (postcopy_start(s)) {
>>               error_report("%s: postcopy failed to start", __func__);
>>           }
>> diff --git a/migration/savevm.c b/migration/savevm.c
>> index 2740defdf0..7a94deda3b 100644
>> --- a/migration/savevm.c
>> +++ b/migration/savevm.c
>> @@ -2504,6 +2504,39 @@ static int loadvm_process_command(QEMUFile *f)
>>       return 0;
>>   }
>>
>> +static int qemu_loadvm_initial_data_loaded_ack(MigrationIncomingState *mis)
>> +{
>> +    SaveStateEntry *se;
>> +    int ret;
>> +
>> +    if (!mis->initial_data_enabled || mis->initial_data_loaded_ack_sent) {
>> +        return 0;
>> +    }
>> +
>> +    QTAILQ_FOREACH(se, &savevm_state.handlers, entry) {
>> +        if (!se->ops || !se->ops->initial_data_loaded) {
>> +            continue;
>> +        }
>> +
>> +        if (!se->ops->is_initial_data_active ||
>> +            !se->ops->is_initial_data_active(se->opaque)) {
>> +            continue;
>> +        }
>> +
>> +        if (!se->ops->initial_data_loaded(se->opaque)) {
>> +            return 0;
>> +        }
>> +    }
>> +
>> +    ret = migrate_send_rp_initial_data_loaded_ack(mis);
>> +    if (!ret) {
>> +        mis->initial_data_loaded_ack_sent = true;
>> +        trace_loadvm_initial_data_loaded_acked();
>> +    }
>> +
>> +    return ret;
>> +}
>> +
>>   /*
>>    * Read a footer off the wire and check that it matches the expected section
>>    *
>> @@ -2826,6 +2859,12 @@ retry:
>>               if (ret < 0) {
>>                   goto out;
>>               }
>> +
>> +            ret = qemu_loadvm_initial_data_loaded_ack(mis);
>> +            if (ret < 0) {
>> +                goto out;
>> +            }
> This is slightly hacky - it gets called for every END section.
>
> Please consider something cleaner, e.g., feel free to consider a
> notification mechanism I mentioned in my reply to the cover letter, so it's
> called only if the device is ready for switchover (no matter what interface
> it'll use).

Sure. Your notification suggestion sounds much better. I will use it.

Thanks.

>> +
>>               break;
>>           case QEMU_VM_COMMAND:
>>               ret = loadvm_process_command(f);
>> diff --git a/migration/trace-events b/migration/trace-events
>> index 21ae471126..a0e1d3b2fd 100644
>> --- a/migration/trace-events
>> +++ b/migration/trace-events
>> @@ -24,6 +24,7 @@ loadvm_postcopy_ram_handle_discard_header(const char *ramid, uint16_t len) "%s:
>>   loadvm_process_command(const char *s, uint16_t len) "com=%s len=%d"
>>   loadvm_process_command_ping(uint32_t val) "0x%x"
>>   loadvm_handle_initial_data_enable(uint8_t general_enable, const char *idstr, int instance_id) "general_enable=%u, idstr=%s, instance_id=%u"
>> +loadvm_initial_data_loaded_acked(void) ""
>>   postcopy_ram_listen_thread_exit(void) ""
>>   postcopy_ram_listen_thread_start(void) ""
>>   qemu_savevm_send_postcopy_advise(void) ""
>> @@ -182,6 +183,7 @@ source_return_path_thread_loop_top(void) ""
>>   source_return_path_thread_pong(uint32_t val) "0x%x"
>>   source_return_path_thread_shut(uint32_t val) "0x%x"
>>   source_return_path_thread_resume_ack(uint32_t v) "%"PRIu32
>> +source_return_path_thread_initial_data_loaded_ack(void) ""
>>   migration_thread_low_pending(uint64_t pending) "%" PRIu64
>>   migrate_transferred(uint64_t tranferred, uint64_t time_spent, uint64_t bandwidth, uint64_t size) "transferred %" PRIu64 " time_spent %" PRIu64 " bandwidth %" PRIu64 " max_size %" PRId64
>>   process_incoming_migration_co_end(int ret, int ps) "ret=%d postcopy-state=%d"
>> --
>> 2.26.3
>>
>>
> --
> Peter Xu
>