seamless migration: migration completion on the destination side

Tracking the channels that wait for migration data. If there
is a new migration process pending, when all the channels have
restored their state, we begin the new migration.

8 files changed, 202 insertions, 25 deletions

diff --git a/server/main_channel.c b/server/main_channel.c
index 63989657..e06bb05e 100644
--- a/server/main_channel.c
+++ b/server/main_channel.c
@@ -837,10 +837,14 @@ void main_channel_client_handle_migrate_end(MainChannelClient *mcc)
                    "client does not support semi-seamless migration");
             return;
     }
-    red_client_migrate_complete(mcc->base.client);
-    if (mcc->mig_wait_prev_complete) {
+    red_client_semi_seamless_migrate_complete(mcc->base.client);
+}
 
+void main_channel_migrate_dst_complete(MainChannelClient *mcc)
+{
+    if (mcc->mig_wait_prev_complete) {
         if (mcc->mig_wait_prev_try_seamless) {
+            spice_assert(mcc->base.channel->clients_num == 1);
             red_channel_client_pipe_add_type(&mcc->base, SPICE_MSG_MAIN_MIGRATE_BEGIN_SEAMLESS);
         } else {
             red_channel_client_pipe_add_type(&mcc->base, SPICE_MSG_MAIN_MIGRATE_BEGIN);
@@ -849,6 +853,7 @@ void main_channel_client_handle_migrate_end(MainChannelClient *mcc)
         mcc->mig_wait_prev_complete = FALSE;
     }
 }
+
 static int main_channel_handle_parsed(RedChannelClient *rcc, uint32_t size, uint16_t type,
                                       void *message)
 {
@@ -1248,7 +1253,7 @@ void main_channel_migrate_cancel_wait(MainChannel *main_chan)
     main_chan->num_clients_mig_wait = 0;
 }
 
-int main_channel_migrate_complete(MainChannel *main_chan, int success)
+int main_channel_migrate_src_complete(MainChannel *main_chan, int success)
 {
     RingItem *client_link;
     int semi_seamless_count = 0;
diff --git a/server/main_channel.h b/server/main_channel.h
index 40d2215a..8cfe62b5 100644
--- a/server/main_channel.h
+++ b/server/main_channel.h
@@ -104,7 +104,8 @@ int main_channel_migrate_connect(MainChannel *main_channel, RedsMigSpice *mig_ta
                                  int try_seamless);
 void main_channel_migrate_cancel_wait(MainChannel *main_chan);
 /* returns the number of clients for which SPICE_MSG_MAIN_MIGRATE_END was sent*/
-int main_channel_migrate_complete(MainChannel *main_chan, int success);
+int main_channel_migrate_src_complete(MainChannel *main_chan, int success);
+void main_channel_migrate_dst_complete(MainChannelClient *mcc);
 void main_channel_push_name(MainChannelClient *mcc, const char *name);
 void main_channel_push_uuid(MainChannelClient *mcc, const uint8_t uuid[16]);
 
diff --git a/server/main_dispatcher.c b/server/main_dispatcher.c
index f5b8b4c4..1126ec06 100644
--- a/server/main_dispatcher.c
+++ b/server/main_dispatcher.c
@@ -7,6 +7,8 @@
 #include "red_common.h"
 #include "dispatcher.h"
 #include "main_dispatcher.h"
+#include "red_channel.h"
+#include "reds.h"
 
 /*
  * Main Dispatcher
@@ -37,6 +39,7 @@ MainDispatcher main_dispatcher;
 
 enum {
     MAIN_DISPATCHER_CHANNEL_EVENT = 0,
+    MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
 
     MAIN_DISPATCHER_NUM_MESSAGES
 };
@@ -46,6 +49,10 @@ typedef struct MainDispatcherChannelEventMessage {
     SpiceChannelEventInfo *info;
 } MainDispatcherChannelEventMessage;
 
+typedef struct MainDispatcherMigrateSeamlessDstCompleteMessage {
+    RedClient *client;
+} MainDispatcherMigrateSeamlessDstCompleteMessage;
+
 /* channel_event - calls core->channel_event, must be done in main thread */
 static void main_dispatcher_self_handle_channel_event(
                                                 int event,
@@ -80,6 +87,28 @@ void main_dispatcher_channel_event(int event, SpiceChannelEventInfo *info)
                             &msg);
 }
 
+
+static void main_dispatcher_handle_migrate_complete(void *opaque,
+                                                    void *payload)
+{
+    MainDispatcherMigrateSeamlessDstCompleteMessage *mig_complete = payload;
+
+    reds_on_client_seamless_migrate_complete(mig_complete->client);
+}
+
+void main_dispatcher_seamless_migrate_dst_complete(RedClient *client)
+{
+    MainDispatcherMigrateSeamlessDstCompleteMessage msg;
+
+    if (pthread_self() == main_dispatcher.base.self) {
+        reds_on_client_seamless_migrate_complete(client);
+        return;
+    }
+
+    msg.client = client;
+    dispatcher_send_message(&main_dispatcher.base, MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
+                            &msg);
+}
 static void dispatcher_handle_read(int fd, int event, void *opaque)
 {
     Dispatcher *dispatcher = opaque;
@@ -97,4 +126,7 @@ void main_dispatcher_init(SpiceCoreInterface *core)
     dispatcher_register_handler(&main_dispatcher.base, MAIN_DISPATCHER_CHANNEL_EVENT,
                                 main_dispatcher_handle_channel_event,
                                 sizeof(MainDispatcherChannelEventMessage), 0 /* no ack */);
+    dispatcher_register_handler(&main_dispatcher.base, MAIN_DISPATCHER_MIGRATE_SEAMLESS_DST_COMPLETE,
+                                main_dispatcher_handle_migrate_complete,
+                                sizeof(MainDispatcherMigrateSeamlessDstCompleteMessage), 0 /* no ack */);
 }
diff --git a/server/main_dispatcher.h b/server/main_dispatcher.h
index 2c201c75..ec4a6b44 100644
--- a/server/main_dispatcher.h
+++ b/server/main_dispatcher.h
@@ -4,6 +4,7 @@
 #include <spice.h>
 
 void main_dispatcher_channel_event(int event, SpiceChannelEventInfo *info);
+void main_dispatcher_seamless_migrate_dst_complete(RedClient *client);
 void main_dispatcher_init(SpiceCoreInterface *core);
 
 #endif //MAIN_DISPATCHER_H
diff --git a/server/red_channel.c b/server/red_channel.c
index d715ce8c..65ef2da5 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -36,6 +36,7 @@
 #include "stat.h"
 #include "red_channel.h"
 #include "reds.h"
+#include "main_dispatcher.h"
 
 static void red_channel_client_event(int fd, int event, void *data);
 static void red_client_add_channel(RedClient *client, RedChannelClient *rcc);
@@ -688,6 +689,45 @@ error:
     return NULL;
 }
 
+static void red_channel_client_seamless_migration_done(RedChannelClient *rcc)
+{
+    rcc->wait_migrate_data = FALSE;
+
+    pthread_mutex_lock(&rcc->client->lock);
+    rcc->client->num_migrated_channels--;
+
+    /* we assume we always have at least one channel who has migration data transfer,
+     * otherwise, this flag will never be set back to FALSE*/
+    if (!rcc->client->num_migrated_channels) {
+        rcc->client->during_target_migrate = FALSE;
+        rcc->client->seamless_migrate = FALSE;
+        /* migration completion might have been triggered from a different thread
+         * than the main thread */
+        main_dispatcher_seamless_migrate_dst_complete(rcc->client);
+    }
+    pthread_mutex_unlock(&rcc->client->lock);
+}
+
+int red_channel_client_waits_for_migrate_data(RedChannelClient *rcc)
+{
+    return rcc->wait_migrate_data;
+}
+
+int red_channel_waits_for_migrate_data(RedChannel *channel)
+{
+    RedChannelClient *rcc;
+    if (!red_channel_is_connected(channel)) {
+        return FALSE;
+    }
+
+    if (channel->clients_num > 1) {
+        return FALSE;
+    }
+    spice_assert(channel->clients_num == 1);
+    rcc = SPICE_CONTAINEROF(ring_get_head(&channel->clients), RedChannelClient, channel_link);
+    return red_channel_client_waits_for_migrate_data(rcc);
+}
+
 static void red_channel_client_default_connect(RedChannel *channel, RedClient *client,
                                                RedsStream *stream,
                                                int migration,
@@ -1085,13 +1125,21 @@ static void red_channel_handle_migrate_flush_mark(RedChannelClient *rcc)
 // So need to make all the handlers work with per channel/client data (what data exactly?)
 static void red_channel_handle_migrate_data(RedChannelClient *rcc, uint32_t size, void *message)
 {
+    spice_debug("channel type %d id %d rcc %p size %u",
+                rcc->channel->type, rcc->channel->id, rcc, size);
     if (!rcc->channel->channel_cbs.handle_migrate_data) {
         return;
     }
-    spice_assert(red_channel_client_get_message_serial(rcc) == 0);
-    red_channel_client_set_message_serial(rcc,
-        rcc->channel->channel_cbs.handle_migrate_data_get_serial(rcc, size, message));
+    if (!red_channel_client_waits_for_migrate_data(rcc)) {
+        spice_error("unexcpected");
+        return;
+    }
+    if (rcc->channel->channel_cbs.handle_migrate_data_get_serial) {
+        red_channel_client_set_message_serial(rcc,
+            rcc->channel->channel_cbs.handle_migrate_data_get_serial(rcc, size, message));
+    }
     rcc->channel->channel_cbs.handle_migrate_data(rcc, size, message);
+    red_channel_client_seamless_migration_done(rcc);
 }
 
 int red_channel_client_handle_message(RedChannelClient *rcc, uint32_t size,
@@ -1555,11 +1603,38 @@ RedClient *red_client_new(int migrated)
     ring_init(&client->channels);
     pthread_mutex_init(&client->lock, NULL);
     client->thread_id = pthread_self();
-    client->migrated = migrated;
+    client->during_target_migrate = migrated;
 
     return client;
 }
 
+/* client mutex should be locked before this call */
+static void red_channel_client_set_migration_seamless(RedChannelClient *rcc)
+{
+    spice_assert(rcc->client->during_target_migrate && rcc->client->seamless_migrate);
+
+    if (rcc->channel->migration_flags & SPICE_MIGRATE_NEED_DATA_TRANSFER) {
+        rcc->wait_migrate_data = TRUE;
+        rcc->client->num_migrated_channels++;
+    }
+    spice_debug("channel type %d id %d rcc %p wait data %d", rcc->channel->type, rcc->channel->id, rcc,
+        rcc->wait_migrate_data);
+}
+
+void red_client_set_migration_seamless(RedClient *client) // dest
+{
+    RingItem *link;
+    pthread_mutex_lock(&client->lock);
+    client->seamless_migrate = TRUE;
+    /* update channel clients that got connected before the migration
+     * type was set. red_client_add_channel will handle newer channel clients */
+    RING_FOREACH(link, &client->channels) {
+        RedChannelClient *rcc = SPICE_CONTAINEROF(link, RedChannelClient, client_link);
+        red_channel_client_set_migration_seamless(rcc);
+    }
+    pthread_mutex_unlock(&client->lock);
+}
+
 void red_client_migrate(RedClient *client)
 {
     RingItem *link, *next;
@@ -1625,6 +1700,9 @@ static void red_client_add_channel(RedClient *client, RedChannelClient *rcc)
 {
     spice_assert(rcc && client);
     ring_add(&client->channels, &rcc->client_link);
+    if (client->during_target_migrate && client->seamless_migrate) {
+        red_channel_client_set_migration_seamless(rcc);
+    }
     client->channels_num++;
 }
 
@@ -1636,16 +1714,27 @@ void red_client_set_main(RedClient *client, MainChannelClient *mcc) {
     client->mcc = mcc;
 }
 
-void red_client_migrate_complete(RedClient *client)
+void red_client_semi_seamless_migrate_complete(RedClient *client)
 {
-    spice_assert(client->migrated);
-    client->migrated = FALSE;
-    reds_on_client_migrate_complete(client);
+    pthread_mutex_lock(&client->lock);
+    if (!client->during_target_migrate || client->seamless_migrate) {
+        spice_error("unexpected");
+        pthread_mutex_unlock(&client->lock);
+        return;
+    }
+    client->during_target_migrate = FALSE;
+    pthread_mutex_unlock(&client->lock);
+    reds_on_client_semi_seamless_migrate_complete(client);
 }
 
+/* should be called only from the main thread */
 int red_client_during_migrate_at_target(RedClient *client)
 {
-    return client->migrated;
+    int ret;
+    pthread_mutex_lock(&client->lock);
+    ret = client->during_target_migrate;
+    pthread_mutex_unlock(&client->lock);
+    return ret;
 }
 
 /*
diff --git a/server/red_channel.h b/server/red_channel.h
index 35d11a61..8c8e1c8b 100644
--- a/server/red_channel.h
+++ b/server/red_channel.h
@@ -267,6 +267,8 @@ struct RedChannelClient {
     RedChannelCapabilities remote_caps;
     int is_mini_header;
     int destroying;
+
+    int wait_migrate_data;
     int wait_migrate_flush_mark;
 };
 
@@ -357,6 +359,11 @@ int red_channel_is_connected(RedChannel *channel);
 int red_channel_client_is_connected(RedChannelClient *rcc);
 
 void red_channel_client_default_migrate(RedChannelClient *rcc);
+int red_channel_client_waits_for_migrate_data(RedChannelClient *rcc);
+/* seamless migration is supported for only one client. This routine
+ * checks if the only channel client associated with channel is
+ * waiting for migration data */
+int red_channel_waits_for_migrate_data(RedChannel *channel);
 
 /*
  * the disconnect callback is called from the channel's thread,
@@ -513,14 +520,31 @@ struct RedClient {
     pthread_t thread_id;
 
     int disconnecting;
-    int migrated;
+    /* Note that while semi-seamless migration is conducted by the main thread, seamless migration
+     * involves all channels, and thus the related varaibles can be accessed from different
+     * threads */
+    int during_target_migrate; /* if seamless=TRUE, migration_target is turned off when all
+                                  the clients received their migration data. Otherwise (semi-seamless),
+                                  it is turned off, when red_client_semi_seamless_migrate_complete
+                                  is called */
+    int seamless_migrate;
+    int num_migrated_channels; /* for seamless - number of channels that wait for migrate data*/
 };
 
 RedClient *red_client_new(int migrated);
+
 MainChannelClient *red_client_get_main(RedClient *client);
 // main should be set once before all the other channels are created
 void red_client_set_main(RedClient *client, MainChannelClient *mcc);
-void red_client_migrate_complete(RedClient *client);
+
+/* called when the migration handshake results in seamless migration (dst side).
+ * By default we assume semi-seamless */
+void red_client_set_migration_seamless(RedClient *client);
+void red_client_semi_seamless_migrate_complete(RedClient *client); /* dst side */
+/* TRUE if the migration is seamless and there are still channels that wait from migration data.
+ * Or, during semi-seamless migration, and the main channel still waits for MIGRATE_END
+ * from the client.
+ * Note: Call it only from the main thread */
 int red_client_during_migrate_at_target(RedClient *client);
 
 void red_client_migrate(RedClient *client);
diff --git a/server/reds.c b/server/reds.c
index 19e72bd0..c957b5da 100644
--- a/server/reds.c
+++ b/server/reds.c
@@ -1440,6 +1440,21 @@ static void reds_mig_target_client_disconnect_all(void)
     }
 }
 
+static int reds_find_client(RedClient *client)
+{
+    RingItem *item;
+
+    RING_FOREACH(item, &reds->clients) {
+        RedClient *list_client;
+
+        list_client = SPICE_CONTAINEROF(item, RedClient, link);
+        if (list_client == client) {
+            return TRUE;
+        }
+    }
+    return FALSE;
+}
+
 // TODO: now that main is a separate channel this should
 // actually be joined with reds_handle_other_links, become reds_handle_link
 static void reds_handle_main_link(RedLinkInfo *link)
@@ -1625,18 +1640,26 @@ int reds_on_migrate_dst_set_seamless(MainChannelClient *mcc, uint32_t src_versio
     if (reds->allow_multiple_clients  || src_version > SPICE_MIGRATION_PROTOCOL_VERSION) {
         reds->dst_do_seamless_migrate = FALSE;
     } else {
-        RedClient *client;
+        RedChannelClient *rcc = main_channel_client_get_base(mcc);
 
-        client = main_channel_client_get_base(mcc)->client;
-        reds->dst_do_seamless_migrate = reds_link_mig_target_channels(client);
+        red_client_set_migration_seamless(rcc->client);
         /* linking all the channels that have been connected before migration handshake */
-        reds->dst_do_seamless_migrate = reds_link_mig_target_channels(client);
+        reds->dst_do_seamless_migrate = reds_link_mig_target_channels(rcc->client);
     }
     return reds->dst_do_seamless_migrate;
 }
 
-/* semi seamless */
-void reds_on_client_migrate_complete(RedClient *client)
+void reds_on_client_seamless_migrate_complete(RedClient *client)
+{
+    spice_debug(NULL);
+    if (!reds_find_client(client)) {
+        spice_info("client no longer exists");
+        return;
+    }
+    main_channel_migrate_dst_complete(red_client_get_main(client));
+}
+
+void reds_on_client_semi_seamless_migrate_complete(RedClient *client)
 {
     MainChannelClient *mcc;
 
@@ -1649,6 +1672,7 @@ void reds_on_client_migrate_complete(RedClient *client)
                            reds_get_mm_time() - MM_TIME_DELTA,
                            red_dispatcher_qxl_ram_size());
     reds_link_mig_target_channels(client);
+    main_channel_migrate_dst_complete(mcc);
 }
 
 static void reds_handle_other_links(RedLinkInfo *link)
@@ -3202,7 +3226,7 @@ static void reds_mig_finished(int completed)
     if (reds->src_do_seamless_migrate && completed) {
         reds_migrate_channels_seamless();
     } else {
-        main_channel_migrate_complete(reds->main_channel, completed);
+        main_channel_migrate_src_complete(reds->main_channel, completed);
     }
 
     if (completed) {
@@ -4067,8 +4091,8 @@ SPICE_GNUC_VISIBLE int spice_server_migrate_connect(SpiceServer *s, const char*
     spice_assert(reds == s);
 
     if (reds->expect_migrate) {
-        spice_error("consecutive calls without migration. Canceling previous call");
-        main_channel_migrate_complete(reds->main_channel, FALSE);
+        spice_info("consecutive calls without migration. Canceling previous call");
+        main_channel_migrate_src_complete(reds->main_channel, FALSE);
     }
 
     sif = SPICE_CONTAINEROF(migration_interface->base.sif, SpiceMigrateInterface, base);
diff --git a/server/reds.h b/server/reds.h
index 71a0173f..779d0dbe 100644
--- a/server/reds.h
+++ b/server/reds.h
@@ -157,7 +157,8 @@ void reds_on_main_mouse_mode_request(void *message, size_t size);
 /* migration dest side: returns whether it can support seamless migration
  * with the given src migration protocol version */
 int reds_on_migrate_dst_set_seamless(MainChannelClient *mcc, uint32_t src_version);
-void reds_on_client_migrate_complete(RedClient *client);
+void reds_on_client_semi_seamless_migrate_complete(RedClient *client);
+void reds_on_client_seamless_migrate_complete(RedClient *client);
 void reds_on_char_device_state_destroy(SpiceCharDeviceState *dev);
 
 #endif