[Qemu-devel] [RFC PATCH 4/5] migration: replace migration state change notifier with async notifiers

[Yonit Halperin]

The patch replaces the existing state change notifier list, with two
explicit notifier lists for migration start and migration end. Unlike
the previous notifier list, the current notifications take place before
the actual status change, and also allow async handlers. Hence, it is
possible to delay the migration progress, till the async handlers have
completed.

Note that this patch still leaves the registered notifier handlers synchronous, 
i.e.,
they call the notifier completion callback immediately.

Signed-off-by: Yonit Halperin <address@hidden>
---
 migration.c     |   84 +++++++++++++++++++++++++++++++++++++-----------------
 migration.h     |    8 ++++-
 ui/spice-core.c |   31 ++++++++++++++------
 3 files changed, 84 insertions(+), 39 deletions(-)

diff --git a/migration.c b/migration.c
index c86611d..869a8ab 100644
--- a/migration.c
+++ b/migration.c
@@ -51,8 +51,15 @@ enum {
 
 #define MAX_THROTTLE  (32 << 20)      /* Migration speed throttling */
 
-static NotifierList migration_state_notifiers =
-    NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
+static void migrate_start(void *opaque);
+static void migrate_end(void *opaque);
+
+static NotifierList migration_start_notifiers =
+    ASYNC_NOTIFIER_LIST_INITIALIZER(migration_start_notifiers,
+                                    migrate_start, NULL);
+static NotifierList migration_end_notifiers =
+    ASYNC_NOTIFIER_LIST_INITIALIZER(migration_end_notifiers,
+                                    migrate_end, NULL);
 
 /* When we add fault tolerance, we could have several
    migrations at once.  For now we don't need to add
@@ -187,31 +194,44 @@ static int migrate_fd_cleanup(MigrationState *s)
     return ret;
 }
 
-static void migrate_end(MigrationState *s, int end_state)
+static void migrate_notify_end(MigrationState *s, int end_state)
+{
+    bool migrate_success = (end_state == MIG_STATE_COMPLETED);
+
+    if (!s->end_was_notified) {
+        s->end_state = end_state;
+        migration_end_notifiers.complete_cb = migrate_end;
+        migration_end_notifiers.complete_opaque = s;
+        s->end_was_notified = true;
+        notifier_list_notify(&migration_end_notifiers, &migrate_success);
+    }
+}
+
+static void migrate_end(void *opaque)
 {
-    s->state = end_state;
+    MigrationState *s = opaque;
+
+    s->state = s->end_state;
     if (s->state == MIG_STATE_COMPLETED) {
         runstate_set(RUN_STATE_POSTMIGRATE);
     } else if (s->state == MIG_STATE_ERROR && s->start_vm_in_error) {
         vm_start();
     }
-    notifier_list_notify(&migration_state_notifiers, s);
 }
 
 void migrate_fd_error(MigrationState *s)
 {
-    DPRINTF("setting error state\n");
     migrate_fd_cleanup(s);
-    migrate_end(s, MIG_STATE_ERROR);
+    migrate_notify_end(s, MIG_STATE_ERROR);
 }
 
 static void migrate_fd_completed(MigrationState *s)
 {
     DPRINTF("setting completed state\n");
     if (migrate_fd_cleanup(s) < 0) {
-        migrate_end(s, MIG_STATE_ERROR);
+        migrate_notify_end(s, MIG_STATE_ERROR);
     } else {
-        migrate_end(s, MIG_STATE_COMPLETED);
+        migrate_notify_end(s, MIG_STATE_COMPLETED);
     }
 }
 
@@ -281,14 +301,16 @@ static void migrate_fd_put_ready(void *opaque)
 
 static void migrate_fd_cancel(MigrationState *s)
 {
-    if (s->state != MIG_STATE_ACTIVE)
+    if (s->state != MIG_STATE_ACTIVE ||
+        notifier_list_async_waiting(&migration_end_notifiers)) {
         return;
+    }
 
     DPRINTF("cancelling migration\n");
 
     qemu_savevm_state_cancel(s->file);
     migrate_fd_cleanup(s);
-    migrate_end(s, MIG_STATE_CANCELLED);
+    migrate_notify_end(s, MIG_STATE_CANCELLED);
 }
 
 static void migrate_fd_wait_for_unfreeze(void *opaque)
@@ -322,14 +344,19 @@ static int migrate_fd_close(void *opaque)
     return s->close(s);
 }
 
-void add_migration_state_change_notifier(Notifier *notify)
+void migration_add_start_notifier(AsyncNotifier *notify)
+{
+    notifier_list_add_async(&migration_start_notifiers, notify);
+}
+
+void migration_add_end_notifier(AsyncNotifier *notify)
 {
-    notifier_list_add(&migration_state_notifiers, notify);
+    notifier_list_add_async(&migration_end_notifiers, notify);
 }
 
-void remove_migration_state_change_notifier(Notifier *notify)
+void migration_remove_state_notifer(AsyncNotifier *notifier)
 {
-    notifier_remove(&notify->base);
+    notifier_remove(&notifier->base);
 }
 
 bool migration_is_active(MigrationState *s)
@@ -401,13 +428,15 @@ void migrate_del_blocker(Error *reason)
     migration_blockers = g_slist_remove(migration_blockers, reason);
 }
 
-static void migrate_start(MigrationState *s, Error **errp)
+static void migrate_start(void *opaque)
 {
+    MigrationState *s = opaque;
     int ret;
+    Error *err = NULL;
 
     switch (s->protocol) {
     case MIGRATION_PROTOCOL_TCP:
-        ret = tcp_start_outgoing_migration(s, s->protocol_param, errp);
+        ret = tcp_start_outgoing_migration(s, s->protocol_param, &err);
         break;
 #if !defined(WIN32)
     case MIGRATION_PROTOCOL_EXEC:
@@ -425,17 +454,18 @@ static void migrate_start(MigrationState *s, Error **errp)
     }
 
     g_free(s->protocol_param);
-    s->protocol_param = NULL;
-
     if (ret < 0) {
-        if (!error_is_set(errp)) {
-            DPRINTF("migration failed: %s\n", strerror(-ret));
-            /* FIXME: we should return meaningful errors */
-            error_set(errp, QERR_UNDEFINED_ERROR);
+        DPRINTF("migration failed: %s\n", strerror(-ret));
+        if (error_is_set(&err)) {
+            fprintf(stderr, "migrate: %s\n", error_get_pretty(err));
         }
+        /* if migration state is not ACTIVE, another migration can start
+         * before all the registered async notifieres completed. In this case
+         * notifier_list_notify, for migration start notification,
+         * will handle not waiting for the previous notification to complete */
+        migrate_notify_end(s, MIG_STATE_ERROR);
         return;
     }
-    notifier_list_notify(&migration_state_notifiers, s);
 }
 
 void qmp_migrate(const char *uri, bool has_blk, bool blk,
@@ -475,9 +505,9 @@ void qmp_migrate(const char *uri, bool has_blk, bool blk,
         return;
     }
     s = migrate_init(migrate_protocol, p, blk, inc);
-
-    migrate_start(s, errp);
-
+    migration_start_notifiers.complete_cb = migrate_start;
+    migration_start_notifiers.complete_opaque = s;
+    notifier_list_notify(&migration_start_notifiers, NULL);
 }
 
 void qmp_migrate_cancel(Error **errp)
diff --git a/migration.h b/migration.h
index 6a0f49f..eeed6ec 100644
--- a/migration.h
+++ b/migration.h
@@ -36,6 +36,8 @@ struct MigrationState
     int protocol;
     char *protocol_param;
     bool start_vm_in_error;
+    int end_state;
+    bool end_was_notified;
 };
 
 void process_incoming_migration(QEMUFile *f);
@@ -69,8 +71,10 @@ void migrate_fd_error(MigrationState *s);
 
 void migrate_fd_connect(MigrationState *s);
 
-void add_migration_state_change_notifier(Notifier *notify);
-void remove_migration_state_change_notifier(Notifier *notify);
+void migration_add_start_notifier(AsyncNotifier *notify);
+/* notification also contains whether the migration was successful */
+void migration_add_end_notifier(AsyncNotifier *notify);
+void migration_remove_state_notifer(AsyncNotifier *notifier);
 bool migration_is_active(MigrationState *);
 bool migration_has_finished(MigrationState *);
 bool migration_has_failed(MigrationState *);
diff --git a/ui/spice-core.c b/ui/spice-core.c
index 4fc48f8..d85c212 100644
--- a/ui/spice-core.c
+++ b/ui/spice-core.c
@@ -41,7 +41,8 @@
 /* core bits */
 
 static SpiceServer *spice_server;
-static Notifier migration_state;
+static AsyncNotifier migrate_start_notifier;
+static AsyncNotifier migrate_end_notifier;
 static const char *auth = "spice";
 static char *auth_passwd;
 static time_t auth_expires = TIME_MAX;
@@ -476,23 +477,31 @@ SpiceInfo *qmp_query_spice(Error **errp)
     return info;
 }
 
-static void migration_state_notifier(Notifier *notifier, void *data)
+static void migrate_start_notify_func(AsyncNotifier *notifier, void *data,
+                                      NotifiedCompletionFunc *complete_cb,
+                                      void *cb_data)
 {
-    MigrationState *s = data;
-
-    if (migration_is_active(s)) {
 #ifdef SPICE_INTERFACE_MIGRATION
-        spice_server_migrate_start(spice_server);
+    spice_server_migrate_start(spice_server);
 #endif
-    } else if (migration_has_finished(s)) {
+    complete_cb(notifier, cb_data);
+}
+
+static void migrate_end_notify_func(AsyncNotifier *notifier, void *data,
+                                    NotifiedCompletionFunc *complete_cb,
+                                    void *cb_data)
+{
+    bool success_end = *(bool *)data;
+    if (success_end) {
 #ifndef SPICE_INTERFACE_MIGRATION
         spice_server_migrate_switch(spice_server);
 #else
         spice_server_migrate_end(spice_server, true);
-    } else if (migration_has_failed(s)) {
+    } else {
         spice_server_migrate_end(spice_server, false);
 #endif
     }
+    complete_cb(notifier, cb_data);
 }
 
 int qemu_spice_migrate_info(const char *hostname, int port, int tls_port,
@@ -707,8 +716,10 @@ void qemu_spice_init(void)
     };
     using_spice = 1;
 
-    migration_state.notify = migration_state_notifier;
-    add_migration_state_change_notifier(&migration_state);
+    migrate_start_notifier.notify_async = migrate_start_notify_func;
+    migration_add_start_notifier(&migrate_start_notifier);
+    migrate_end_notifier.notify_async = migrate_end_notify_func;
+    migration_add_end_notifier(&migrate_end_notifier);
 #ifdef SPICE_INTERFACE_MIGRATION
     spice_migrate.sin.base.sif = &migrate_interface.base;
     spice_migrate.connect_complete.cb = NULL;
-- 
1.7.7.6