Re: [PATCH v5 17/36] migration: Add save_live_complete_precopy_thread handler

[Peter Xu <peterx@redhat.com>]

On Tue, Mar 04, 2025 at 10:50:29PM +0100, Maciej S. Szmigiero wrote:
> On 26.02.2025 17:43, Peter Xu wrote:
> > On Wed, Feb 19, 2025 at 09:33:59PM +0100, Maciej S. Szmigiero wrote:
> > > From: "Maciej S. Szmigiero" <maciej.szmigiero@oracle.com>
> > > 
> > > This SaveVMHandler helps device provide its own asynchronous transmission
> > > of the remaining data at the end of a precopy phase via multifd channels,
> > > in parallel with the transfer done by save_live_complete_precopy handlers.
> > > 
> > > These threads are launched only when multifd device state transfer is
> > > supported.
> > > 
> > > Management of these threads in done in the multifd migration code,
> > > wrapping them in the generic thread pool.
> > > 
> > > Signed-off-by: Maciej S. Szmigiero <maciej.szmigiero@oracle.com>
> > > ---
> > >   include/migration/misc.h         | 17 +++++++
> > >   include/migration/register.h     | 19 +++++++
> > >   include/qemu/typedefs.h          |  3 ++
> > >   migration/multifd-device-state.c | 85 ++++++++++++++++++++++++++++++++
> > >   migration/savevm.c               | 35 ++++++++++++-
> > >   5 files changed, 158 insertions(+), 1 deletion(-)
> > > 
> > > diff --git a/include/migration/misc.h b/include/migration/misc.h
> > > index 273ebfca6256..8fd36eba1da7 100644
> > > --- a/include/migration/misc.h
> > > +++ b/include/migration/misc.h
> > > @@ -119,8 +119,25 @@ bool migrate_uri_parse(const char *uri, MigrationChannel **channel,
> > >                          Error **errp);
> > >   /* migration/multifd-device-state.c */
> > > +typedef struct SaveLiveCompletePrecopyThreadData {
> > > +    SaveLiveCompletePrecopyThreadHandler hdlr;
> > > +    char *idstr;
> > > +    uint32_t instance_id;
> > > +    void *handler_opaque;
> > > +} SaveLiveCompletePrecopyThreadData;
> > > +
> > >   bool multifd_queue_device_state(char *idstr, uint32_t instance_id,
> > >                                   char *data, size_t len);
> > >   bool multifd_device_state_supported(void);
> > > +void
> > > +multifd_spawn_device_state_save_thread(SaveLiveCompletePrecopyThreadHandler hdlr,
> > > +                                       char *idstr, uint32_t instance_id,
> > > +                                       void *opaque);
> > > +
> > > +bool multifd_device_state_save_thread_should_exit(void);
> > > +
> > > +void multifd_abort_device_state_save_threads(void);
> > > +bool multifd_join_device_state_save_threads(void);
> > > +
> > >   #endif
> > > diff --git a/include/migration/register.h b/include/migration/register.h
> > > index 58891aa54b76..c041ce32f2fc 100644
> > > --- a/include/migration/register.h
> > > +++ b/include/migration/register.h
> > > @@ -105,6 +105,25 @@ typedef struct SaveVMHandlers {
> > >        */
> > >       int (*save_live_complete_precopy)(QEMUFile *f, void *opaque);
> > > +    /**
> > > +     * @save_live_complete_precopy_thread (invoked in a separate thread)
> > > +     *
> > > +     * Called at the end of a precopy phase from a separate worker thread
> > > +     * in configurations where multifd device state transfer is supported
> > > +     * in order to perform asynchronous transmission of the remaining data in
> > > +     * parallel with @save_live_complete_precopy handlers.
> > > +     * When postcopy is enabled, devices that support postcopy will skip this
> > > +     * step.
> > > +     *
> > > +     * @d: a #SaveLiveCompletePrecopyThreadData containing parameters that the
> > > +     * handler may need, including this device section idstr and instance_id,
> > > +     * and opaque data pointer passed to register_savevm_live().
> > > +     * @errp: pointer to Error*, to store an error if it happens.
> > > +     *
> > > +     * Returns true to indicate success and false for errors.
> > > +     */
> > > +    SaveLiveCompletePrecopyThreadHandler save_live_complete_precopy_thread;
> > > +
> > >       /* This runs both outside and inside the BQL.  */
> > >       /**
> > > diff --git a/include/qemu/typedefs.h b/include/qemu/typedefs.h
> > > index fd23ff7771b1..42ed4e6be150 100644
> > > --- a/include/qemu/typedefs.h
> > > +++ b/include/qemu/typedefs.h
> > > @@ -108,6 +108,7 @@ typedef struct QString QString;
> > >   typedef struct RAMBlock RAMBlock;
> > >   typedef struct Range Range;
> > >   typedef struct ReservedRegion ReservedRegion;
> > > +typedef struct SaveLiveCompletePrecopyThreadData SaveLiveCompletePrecopyThreadData;
> > >   typedef struct SHPCDevice SHPCDevice;
> > >   typedef struct SSIBus SSIBus;
> > >   typedef struct TCGCPUOps TCGCPUOps;
> > > @@ -133,5 +134,7 @@ typedef struct IRQState *qemu_irq;
> > >   typedef void (*qemu_irq_handler)(void *opaque, int n, int level);
> > >   typedef bool (*MigrationLoadThread)(void *opaque, bool *should_quit,
> > >                                       Error **errp);
> > > +typedef bool (*SaveLiveCompletePrecopyThreadHandler)(SaveLiveCompletePrecopyThreadData *d,
> > > +                                                     Error **errp);
> > >   #endif /* QEMU_TYPEDEFS_H */
> > > diff --git a/migration/multifd-device-state.c b/migration/multifd-device-state.c
> > > index 5de3cf27d6e8..63f021fb8dad 100644
> > > --- a/migration/multifd-device-state.c
> > > +++ b/migration/multifd-device-state.c
> > > @@ -8,7 +8,10 @@
> > >    */
> > >   #include "qemu/osdep.h"
> > > +#include "qapi/error.h"
> > >   #include "qemu/lockable.h"
> > > +#include "block/thread-pool.h"
> > > +#include "migration.h"
> > >   #include "migration/misc.h"
> > >   #include "multifd.h"
> > >   #include "options.h"
> > > @@ -17,6 +20,9 @@ static struct {
> > >       QemuMutex queue_job_mutex;
> > >       MultiFDSendData *send_data;
> > > +
> > > +    ThreadPool *threads;
> > > +    bool threads_abort;
> > >   } *multifd_send_device_state;
> > >   void multifd_device_state_send_setup(void)
> > > @@ -27,10 +33,14 @@ void multifd_device_state_send_setup(void)
> > >       qemu_mutex_init(&multifd_send_device_state->queue_job_mutex);
> > >       multifd_send_device_state->send_data = multifd_send_data_alloc();
> > > +
> > > +    multifd_send_device_state->threads = thread_pool_new();
> > > +    multifd_send_device_state->threads_abort = false;
> > >   }
> > >   void multifd_device_state_send_cleanup(void)
> > >   {
> > > +    g_clear_pointer(&multifd_send_device_state->threads, thread_pool_free);
> > >       g_clear_pointer(&multifd_send_device_state->send_data,
> > >                       multifd_send_data_free);
> > > @@ -115,3 +125,78 @@ bool multifd_device_state_supported(void)
> > >       return migrate_multifd() && !migrate_mapped_ram() &&
> > >           migrate_multifd_compression() == MULTIFD_COMPRESSION_NONE;
> > >   }
> > > +
> > > +static void multifd_device_state_save_thread_data_free(void *opaque)
> > > +{
> > > +    SaveLiveCompletePrecopyThreadData *data = opaque;
> > > +
> > > +    g_clear_pointer(&data->idstr, g_free);
> > > +    g_free(data);
> > > +}
> > > +
> > > +static int multifd_device_state_save_thread(void *opaque)
> > > +{
> > > +    SaveLiveCompletePrecopyThreadData *data = opaque;
> > > +    g_autoptr(Error) local_err = NULL;
> > > +
> > > +    if (!data->hdlr(data, &local_err)) {
> > > +        MigrationState *s = migrate_get_current();
> > > +
> > > +        assert(local_err);
> > > +
> > > +        /*
> > > +         * In case of multiple save threads failing which thread error
> > > +         * return we end setting is purely arbitrary.
> > > +         */
> > > +        migrate_set_error(s, local_err);
> > 
> > Where did you kick off all the threads when one hit error?  I wonder if
> > migrate_set_error() should just set quit flag for everything, but for this
> > series it might be easier to use multifd_abort_device_state_save_threads().
> 
> I've now added call to multifd_abort_device_state_save_threads() if a migration
> error is already set to avoid needlessly waiting for the remaining threads to
> do all of their work.

With that, feel free to take:

Reviewed-by: Peter Xu <peterx@redhat.com>

-- 
Peter Xu