server/red_channel (all): handle MIGRATE_DATA and MIGRATE_FLUSH_DATA

Handling done in red_channel instead of per channel, using call backs
for the channel specific part.
Intended to reduce furthur reliance of channels on RedChannel struct.

The commit makes the code harder to understand because of the artificial
get_serial stuff, should later be fixed by having a joint migration
header with the serial (since all channels pass it).

7 files changed, 169 insertions, 52 deletions

diff --git a/server/inputs_channel.c b/server/inputs_channel.c
index 9ebf0509..8140c04e 100644
--- a/server/inputs_channel.c
+++ b/server/inputs_channel.c
@@ -529,7 +529,10 @@ static void inputs_link(Channel *channel, RedsStream *stream, int migration,
         ,inputs_channel_send_item
         ,inputs_channel_release_pipe_item
         ,inputs_channel_on_incoming_error
-        ,inputs_channel_on_outgoing_error);
+        ,inputs_channel_on_outgoing_error
+        ,NULL
+        ,NULL
+        ,NULL);
     ASSERT(inputs_channel);
     channel->data = inputs_channel;
     inputs_pipe_add_init(inputs_channel);
diff --git a/server/main_channel.c b/server/main_channel.c
index 6acb134c..75accac7 100644
--- a/server/main_channel.c
+++ b/server/main_channel.c
@@ -418,11 +418,31 @@ static void main_channel_marshall_migrate_data_item(SpiceMarshaller *m, int seri
     data->ping_id = ping_id;
 }
 
-static void main_channel_receive_migrate_data(MainChannel *main_chan,
-                                  MainMigrateData *data, uint8_t *end)
+static uint64_t main_channel_handle_migrate_data_get_serial(RedChannel *base,
+    uint32_t size, void *message)
 {
-    red_channel_set_message_serial(&main_chan->base, data->serial);
+    MainMigrateData *data = message;
+
+    if (size < sizeof(*data)) {
+        red_printf("bad message size");
+        return 0;
+    }
+    return data->serial;
+}
+
+static uint64_t main_channel_handle_migrate_data(RedChannel *base,
+    uint32_t size, void *message)
+{
+    MainChannel *main_chan = SPICE_CONTAINEROF(base, MainChannel, base);
+    MainMigrateData *data = message;
+
+    if (size < sizeof(*data)) {
+        red_printf("bad message size");
+        return FALSE;
+    }
     main_chan->ping_id = data->ping_id;
+    reds_on_main_receive_migrate_data(data, ((uint8_t*)message) + size);
+    return TRUE;
 }
 
 void main_channel_push_init(Channel *channel, int connection_id,
@@ -729,15 +749,9 @@ static int main_channel_handle_parsed(RedChannel *channel, uint32_t size, uint16
         break;
     }
     case SPICE_MSGC_MIGRATE_FLUSH_MARK:
-        main_channel_push_migrate_data_item(main_chan);
         break;
     case SPICE_MSGC_MIGRATE_DATA: {
-            MainMigrateData *data = (MainMigrateData *)message;
-            uint8_t *end = ((uint8_t *)message) + size;
-            main_channel_receive_migrate_data(main_chan, data, end);
-            reds_on_main_receive_migrate_data(data, end);
-            break;
-        }
+                }
     case SPICE_MSGC_DISCONNECTING:
         break;
     default:
@@ -772,6 +786,12 @@ static void main_channel_hold_pipe_item(RedChannel *channel, PipeItem *item)
 {
 }
 
+static int main_channel_handle_migrate_flush_mark(RedChannel *base)
+{
+    main_channel_push_migrate_data_item(SPICE_CONTAINEROF(base, MainChannel, base));
+    return TRUE;
+}
+
 static void main_channel_link(Channel *channel, RedsStream *stream, int migration,
                         int num_common_caps, uint32_t *common_caps, int num_caps,
                         uint32_t *caps)
@@ -791,7 +811,10 @@ static void main_channel_link(Channel *channel, RedsStream *stream, int migratio
         ,main_channel_send_item
         ,main_channel_release_pipe_item
         ,main_channel_on_error
-        ,main_channel_on_error);
+        ,main_channel_on_error
+        ,main_channel_handle_migrate_flush_mark
+        ,main_channel_handle_migrate_data
+        ,main_channel_handle_migrate_data_get_serial);
     ASSERT(main_chan);
     channel->data = main_chan;
 }
diff --git a/server/red_channel.c b/server/red_channel.c
index 9028943f..5749dc35 100644
--- a/server/red_channel.c
+++ b/server/red_channel.c
@@ -322,7 +322,10 @@ RedChannel *red_channel_create(int size, RedsStream *stream,
                                channel_release_msg_recv_buf_proc release_recv_buf,
                                channel_hold_pipe_item_proc hold_item,
                                channel_send_pipe_item_proc send_item,
-                               channel_release_pipe_item_proc release_item)
+                               channel_release_pipe_item_proc release_item,
+                               channel_handle_migrate_flush_mark handle_migrate_flush_mark,
+                               channel_handle_migrate_data handle_migrate_data,
+                               channel_handle_migrate_data_get_serial handle_migrate_data_get_serial)
 {
     RedChannel *channel;
 
@@ -336,6 +339,9 @@ RedChannel *red_channel_create(int size, RedsStream *stream,
     channel->send_item = send_item;
     channel->release_item = release_item;
     channel->hold_item = hold_item;
+    channel->handle_migrate_flush_mark = handle_migrate_flush_mark;
+    channel->handle_migrate_data = handle_migrate_data;
+    channel->handle_migrate_data_get_serial = handle_migrate_data_get_serial;
 
     channel->stream = stream;
     channel->core = core;
@@ -406,12 +412,16 @@ RedChannel *red_channel_create_parser(int size, RedsStream *stream,
                                channel_send_pipe_item_proc send_item,
                                channel_release_pipe_item_proc release_item,
                                channel_on_incoming_error_proc incoming_error,
-                               channel_on_outgoing_error_proc outgoing_error)
+                               channel_on_outgoing_error_proc outgoing_error,
+                               channel_handle_migrate_flush_mark handle_migrate_flush_mark,
+                               channel_handle_migrate_data handle_migrate_data,
+                               channel_handle_migrate_data_get_serial handle_migrate_data_get_serial)
 {
     RedChannel *channel = red_channel_create(size, stream,
         core, migrate, handle_acks, config_socket, do_nothing_disconnect,
         do_nothing_handle_message, alloc_recv_buf, release_recv_buf, hold_item,
-        send_item, release_item);
+        send_item, release_item, handle_migrate_flush_mark, handle_migrate_data,
+        handle_migrate_data_get_serial);
 
     if (channel == NULL) {
         return NULL;
@@ -454,6 +464,24 @@ void red_channel_init_outgoing_messages_window(RedChannel *channel)
     red_channel_push(channel);
 }
 
+void red_channel_handle_migrate_flush_mark(RedChannel *channel)
+{
+    if (channel->handle_migrate_flush_mark) {
+        channel->handle_migrate_flush_mark(channel);
+    }
+}
+
+void red_channel_handle_migrate_data(RedChannel *channel, uint32_t size, void *message)
+{
+    if (!channel->handle_migrate_data) {
+        return;
+    }
+    ASSERT(red_channel_get_message_serial(channel) == 0);
+    red_channel_set_message_serial(channel,
+        channel->handle_migrate_data_get_serial(channel, size, message));
+    channel->handle_migrate_data(channel, size, message);
+}
+
 int red_channel_handle_message(RedChannel *channel, uint32_t size,
                                uint16_t type, void *message)
 {
@@ -473,6 +501,12 @@ int red_channel_handle_message(RedChannel *channel, uint32_t size,
         break;
     case SPICE_MSGC_DISCONNECTING:
         break;
+    case SPICE_MSGC_MIGRATE_FLUSH_MARK:
+        red_channel_handle_migrate_flush_mark(channel);
+        break;
+    case SPICE_MSGC_MIGRATE_DATA:
+        red_channel_handle_migrate_data(channel, size, message);
+        break;
     default:
         red_printf("invalid message type %u", type);
         return FALSE;
diff --git a/server/red_channel.h b/server/red_channel.h
index 50e67893..1841de47 100644
--- a/server/red_channel.h
+++ b/server/red_channel.h
@@ -123,6 +123,12 @@ typedef void (*channel_release_pipe_item_proc)(RedChannel *channel,
 typedef void (*channel_on_incoming_error_proc)(RedChannel *channel);
 typedef void (*channel_on_outgoing_error_proc)(RedChannel *channel);
 
+typedef int (*channel_handle_migrate_flush_mark)(RedChannel *channel);
+typedef uint64_t (*channel_handle_migrate_data)(RedChannel *channel,
+                                                uint32_t size, void *message);
+typedef uint64_t (*channel_handle_migrate_data_get_serial)(RedChannel *channel,
+                                            uint32_t size, void *message);
+
 struct RedChannel {
     RedsStream *stream;
     SpiceCoreInterface *core;
@@ -166,6 +172,10 @@ struct RedChannel {
     channel_on_incoming_error_proc on_incoming_error; /* alternative to disconnect */
     channel_on_outgoing_error_proc on_outgoing_error;
     int shut; /* signal channel is to be closed */
+
+    channel_handle_migrate_flush_mark handle_migrate_flush_mark;
+    channel_handle_migrate_data handle_migrate_data;
+    channel_handle_migrate_data_get_serial handle_migrate_data_get_serial;
 };
 
 /* if one of the callbacks should cause disconnect, use red_channel_shutdown and don't
@@ -180,7 +190,10 @@ RedChannel *red_channel_create(int size, RedsStream *stream,
                                channel_release_msg_recv_buf_proc release_recv_buf,
                                channel_hold_pipe_item_proc hold_item,
                                channel_send_pipe_item_proc send_item,
-                               channel_release_pipe_item_proc release_item);
+                               channel_release_pipe_item_proc release_item,
+                               channel_handle_migrate_flush_mark handle_migrate_flush_mark,
+                               channel_handle_migrate_data handle_migrate_data,
+                               channel_handle_migrate_data_get_serial handle_migrate_data_get_serial);
 
 /* alternative constructor, meant for marshaller based (inputs,main) channels,
  * will become default eventually */
@@ -196,7 +209,10 @@ RedChannel *red_channel_create_parser(int size, RedsStream *stream,
                                channel_send_pipe_item_proc send_item,
                                channel_release_pipe_item_proc release_item,
                                channel_on_incoming_error_proc incoming_error,
-                               channel_on_outgoing_error_proc outgoing_error);
+                               channel_on_outgoing_error_proc outgoing_error,
+                               channel_handle_migrate_flush_mark handle_migrate_flush_mark,
+                               channel_handle_migrate_data handle_migrate_data,
+                               channel_handle_migrate_data_get_serial handle_migrate_data_get_serial);
 
 int red_channel_is_connected(RedChannel *channel);
 
diff --git a/server/red_tunnel_worker.c b/server/red_tunnel_worker.c
index a85a1adc..0366290e 100644
--- a/server/red_tunnel_worker.c
+++ b/server/red_tunnel_worker.c
@@ -1739,8 +1739,9 @@ static void __tunnel_channel_fill_socket_migrate_item(TunnelChannel *channel, Re
 }
 
 static void release_migrate_item(TunnelMigrateItem *item);
-static int tunnel_channel_handle_migrate_mark(TunnelChannel *channel)
+static int tunnel_channel_handle_migrate_mark(RedChannel *base)
 {
+    TunnelChannel *channel = SPICE_CONTAINEROF(base, TunnelChannel, base);
     TunnelMigrateItem *migrate_item = NULL;
     TunnelService *service;
     TunnelMigrateServiceItem *mig_service;
@@ -2153,13 +2154,32 @@ static inline void tunnel_channel_activate_migrated_sockets(TunnelChannel *chann
     }
 }
 
-static int tunnel_channel_handle_migrate_data(TunnelChannel *channel,
-                                              TunnelMigrateData *migrate_data)
+static uint64_t tunnel_channel_handle_migrate_data_get_serial(RedChannel *base,
+                                              uint32_t size, void *msg)
 {
+    TunnelMigrateData *migrate_data = msg;
+
+    if (size < sizeof(TunnelMigrateData)
+        || migrate_data->magic != TUNNEL_MIGRATE_DATA_MAGIC
+        || migrate_data->version != TUNNEL_MIGRATE_DATA_VERSION) {
+        return 0;
+    }
+    return migrate_data->message_serial;
+}
+
+static uint64_t tunnel_channel_handle_migrate_data(RedChannel *base,
+                                              uint32_t size, void *msg)
+{
+    TunnelChannel *channel = SPICE_CONTAINEROF(base, TunnelChannel, base);
     TunnelMigrateSocketList *sockets_list;
     TunnelMigrateServicesList *services_list;
+    TunnelMigrateData *migrate_data = msg;
     int i;
 
+    if (size < sizeof(TunnelMigrateData)) {
+        red_printf("bad message size");
+        goto error;
+    }
     if (!channel->expect_migrate_data) {
         red_printf("unexpected");
         goto error;
@@ -2172,9 +2192,6 @@ static int tunnel_channel_handle_migrate_data(TunnelChannel *channel,
         goto error;
     }
 
-    ASSERT(red_channel_get_message_serial(&channel->base) == 0);
-    red_channel_set_message_serial(&channel->base, migrate_data->message_serial);
-
     net_slirp_state_restore(migrate_data->data + migrate_data->slirp_state);
 
     services_list = (TunnelMigrateServicesList *)(migrate_data->data +
@@ -2314,15 +2331,6 @@ static int tunnel_channel_handle_message(RedChannel *channel, SpiceDataHeader *h
 
         return tunnel_channel_handle_socket_token(tunnel_channel, sckt,
                                                   (SpiceMsgcTunnelSocketTokens *)msg);
-    case SPICE_MSGC_MIGRATE_FLUSH_MARK:
-        return tunnel_channel_handle_migrate_mark(tunnel_channel);
-    case SPICE_MSGC_MIGRATE_DATA:
-        if (header->size < sizeof(TunnelMigrateData)) {
-            red_printf("bad message size");
-            free(msg);
-            return FALSE;
-        }
-        return tunnel_channel_handle_migrate_data(tunnel_channel, (TunnelMigrateData *)msg);
     default:
         return red_channel_handle_message(channel, header->size, header->type, msg);
     }
@@ -3425,7 +3433,10 @@ static void handle_tunnel_channel_link(Channel *channel, RedsStream *stream, int
                                             tunnel_channel_release_msg_rcv_buf,
                                             tunnel_channel_hold_pipe_item,
                                             tunnel_channel_send_item,
-                                            tunnel_channel_release_pipe_item);
+                                            tunnel_channel_release_pipe_item,
+                                            tunnel_channel_handle_migrate_mark,
+                                            tunnel_channel_handle_migrate_data,
+                                            tunnel_channel_handle_migrate_data_get_serial);
 
     if (!tunnel_channel) {
         return;
diff --git a/server/red_worker.c b/server/red_worker.c
index 86e512e3..dd89e3fa 100644
--- a/server/red_worker.c
+++ b/server/red_worker.c
@@ -8926,8 +8926,10 @@ static int display_channel_handle_migrate_glz_dictionary(DisplayChannel *channel
                                                              &migrate_info->glz_dict_restore_data));
 }
 
-static int display_channel_handle_migrate_mark(DisplayChannel *channel)
+static int display_channel_handle_migrate_mark(RedChannel *base)
 {
+    DisplayChannel *channel = SPICE_CONTAINEROF(base, DisplayChannel, common.base);
+
     if (!channel->expect_migrate_mark) {
         red_printf("unexpected");
         return FALSE;
@@ -8937,28 +8939,44 @@ static int display_channel_handle_migrate_mark(DisplayChannel *channel)
     return TRUE;
 }
 
-static int display_channel_handle_migrate_data(DisplayChannel *channel, size_t size, void *message)
+static uint64_t display_channel_handle_migrate_data_get_serial(
+                RedChannel *base, uint32_t size, void *message)
+{
+    DisplayChannelMigrateData *migrate_data = message;
+
+    if (size < sizeof(*migrate_data)) {
+        red_printf("bad message size");
+        return 0;
+    }
+    if (migrate_data->magic != DISPLAY_MIGRATE_DATA_MAGIC ||
+        migrate_data->version != DISPLAY_MIGRATE_DATA_VERSION) {
+        red_printf("invalid content");
+        return 0;
+    }
+    return migrate_data->message_serial;
+}
+
+static uint64_t display_channel_handle_migrate_data(RedChannel *base, uint32_t size, void *message)
 {
     DisplayChannelMigrateData *migrate_data;
     int i;
+    DisplayChannel *channel = SPICE_CONTAINEROF(base, DisplayChannel, common.base);
 
-    if (!channel->expect_migrate_data) {
-        red_printf("unexpected");
-        return FALSE;
-    }
-    channel->expect_migrate_data = FALSE;
     if (size < sizeof(*migrate_data)) {
         red_printf("bad message size");
         return FALSE;
     }
     migrate_data = (DisplayChannelMigrateData *)message;
     if (migrate_data->magic != DISPLAY_MIGRATE_DATA_MAGIC ||
-                                            migrate_data->version != DISPLAY_MIGRATE_DATA_VERSION) {
+        migrate_data->version != DISPLAY_MIGRATE_DATA_VERSION) {
         red_printf("invalid content");
         return FALSE;
     }
-    ASSERT(channel->common.base.send_data.serial == 0);
-    channel->common.base.send_data.serial = migrate_data->message_serial;
+    if (!channel->expect_migrate_data) {
+        red_printf("unexpected");
+        return FALSE;
+    }
+    channel->expect_migrate_data = FALSE;
     if (!(channel->pixmap_cache = red_get_pixmap_cache(migrate_data->pixmap_cache_id, -1))) {
         return FALSE;
     }
@@ -8999,10 +9017,6 @@ static int display_channel_handle_message(RedChannel *channel, uint32_t size, ui
         }
         ((DisplayChannel *)channel)->expect_init = FALSE;
         return display_channel_init((DisplayChannel *)channel, (SpiceMsgcDisplayInit *)message);
-    case SPICE_MSGC_MIGRATE_FLUSH_MARK:
-        return display_channel_handle_migrate_mark((DisplayChannel *)channel);
-    case SPICE_MSGC_MIGRATE_DATA:
-        return display_channel_handle_migrate_data((DisplayChannel *)channel, size, message);
     default:
         return red_channel_handle_message(channel, size, type, message);
     }
@@ -9064,7 +9078,10 @@ static RedChannel *__new_channel(RedWorker *worker, int size, uint32_t channel_i
                                  channel_send_pipe_item_proc send_item,
                                  channel_hold_pipe_item_proc hold_item,
                                  channel_release_pipe_item_proc release_item,
-                                 channel_handle_parsed_proc handle_parsed)
+                                 channel_handle_parsed_proc handle_parsed,
+                                 channel_handle_migrate_flush_mark handle_migrate_flush_mark,
+                                 channel_handle_migrate_data handle_migrate_data,
+                                 channel_handle_migrate_data_get_serial handle_migrate_data_get_serial)
 {
     struct epoll_event event;
     RedChannel *channel;
@@ -9081,7 +9098,10 @@ static RedChannel *__new_channel(RedWorker *worker, int size, uint32_t channel_i
                                         send_item,
                                         release_item,
                                         red_channel_default_peer_on_error,
-                                        red_channel_default_peer_on_error);
+                                        red_channel_default_peer_on_error,
+                                        handle_migrate_flush_mark,
+                                        handle_migrate_data,
+                                        handle_migrate_data_get_serial);
     common = (CommonChannel *)channel;
     if (!channel) {
         goto error;
@@ -9186,7 +9206,11 @@ static void handle_new_display_channel(RedWorker *worker, RedsStream *stream, in
                                                             display_channel_send_item,
                                                             display_channel_hold_pipe_item,
                                                             display_channel_release_item,
-                                                            display_channel_handle_message))) {
+                                                            display_channel_handle_message,
+                                                            display_channel_handle_migrate_mark,
+                                                            display_channel_handle_migrate_data,
+                                                            display_channel_handle_migrate_data_get_serial
+                                                            ))) {
         return;
     }
 #ifdef RED_STATISTICS
@@ -9318,7 +9342,10 @@ static void red_connect_cursor(RedWorker *worker, RedsStream *stream, int migrat
                                                    cursor_channel_send_item,
                                                    cursor_channel_hold_pipe_item,
                                                    cursor_channel_release_item,
-                                                   red_channel_handle_message))) {
+                                                   red_channel_handle_message,
+                                                   NULL,
+                                                   NULL,
+                                                   NULL))) {
         return;
     }
 #ifdef RED_STATISTICS
diff --git a/server/smartcard.c b/server/smartcard.c
index 3675cc1a..4c50dbed 100644
--- a/server/smartcard.c
+++ b/server/smartcard.c
@@ -487,7 +487,10 @@ static void smartcard_link(Channel *channel, RedsStream *stream,
                                         smartcard_channel_release_msg_rcv_buf,
                                         smartcard_channel_hold_pipe_item,
                                         smartcard_channel_send_item,
-                                        smartcard_channel_release_pipe_item);
+                                        smartcard_channel_release_pipe_item,
+                                        NULL,
+                                        NULL,
+                                        NULL);
     if (!g_smartcard_channel) {
         return;
     }