Move thread dispatching to RedChannelclientcbs

This allows to directly define
the connect/disconnect/migrate vfuncs in RedCursorChannel and
RedDisplayChannel rather than just having thread dispatching stubs, and
the actual implementation in red-worker.c.
This is only compiling, the RedChannel::thread-dispatcher property does
not exist, there is code to be removed in red-worker.c (the no longer
used CURSOR/DISPLAY dispatching code), and more work needs to be done in
DisplayChannel as some of the needed code lives in red-worker.c :)

5 files changed, 168 insertions, 110 deletions

diff --git a/server/cursor-channel.c b/server/cursor-channel.c
index e39044a8..424d8a29 100644
--- a/server/cursor-channel.c
+++ b/server/cursor-channel.c
@@ -300,6 +300,7 @@ CursorChannel* cursor_channel_new(RedsState *server, QXLInstance *qxl,
                         "id", qxl->id,
                         "migration-flags", 0,
                         "qxl", qxl,
+                        "thread-dispatcher", red_qxl_get_dispatcher(qxl),
                         "handle-acks", TRUE,
                         NULL);
 }
@@ -408,6 +409,7 @@ void cursor_channel_connect(CursorChannel *cursor, RedClient *client, RedsStream
                             RedChannelCapabilities *caps)
 {
     CursorChannelClient *ccc;
+    //CursorChannel *cursor = CURSOR_CHANNEL(channel);
 
     spice_return_if_fail(cursor != NULL);
 
@@ -436,64 +438,14 @@ cursor_channel_finalize(GObject *object)
     G_OBJECT_CLASS(cursor_channel_parent_class)->finalize(object);
 }
 
-static void red_qxl_set_cursor_peer(RedChannel *channel, RedClient *client, RedsStream *stream,
-                                    int migration,
-                                    RedChannelCapabilities *caps)
-{
-    RedWorkerMessageCursorConnect payload = {0,};
-    QXLInstance *qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    Dispatcher *dispatcher = red_qxl_get_dispatcher(qxl);
-    spice_printerr("");
-    // get a reference potentially the main channel can be destroyed in
-    // the main thread causing RedClient to be destroyed before using it
-    payload.client = g_object_ref(client);
-    payload.stream = stream;
-    payload.migration = migration;
-    red_channel_capabilities_init(&payload.caps, caps);
-
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_CURSOR_CONNECT,
-                            &payload);
-}
-
 static void red_qxl_disconnect_cursor_peer(RedChannel *channel, RedChannelClient *rcc)
 {
-    RedWorkerMessageCursorDisconnect payload;
-    QXLInstance *qxl;
-    Dispatcher *dispatcher;
-
-    if (!channel) {
-        return;
-    }
-
-    qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    dispatcher = red_qxl_get_dispatcher(qxl);
-    spice_printerr("");
-    payload.rcc = rcc;
-
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_CURSOR_DISCONNECT,
-                            &payload);
+    red_channel_client_disconnect(rcc);
 }
 
 static void red_qxl_cursor_migrate(RedChannel *channel, RedChannelClient *rcc)
 {
-    RedWorkerMessageCursorMigrate payload;
-    QXLInstance *qxl;
-    Dispatcher *dispatcher;
-    uint32_t type, id;
-
-    if (!channel) {
-        return;
-    }
-    g_object_get(channel, "channel-type", &type, "id", &id, NULL);
-    spice_printerr("channel type %u id %u", type, id);
-    qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    dispatcher = red_qxl_get_dispatcher(qxl);
-    payload.rcc = rcc;
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_CURSOR_MIGRATE,
-                            &payload);
+    cursor_channel_client_migrate(rcc);
 }
 
 static void
@@ -511,7 +463,7 @@ cursor_channel_class_init(CursorChannelClass *klass)
     channel_class->send_item = cursor_channel_send_item;
 
     /* client cbs */
-    channel_class->connect = red_qxl_set_cursor_peer;
+//    channel_class->connect = cursor_channel_connect;
     channel_class->disconnect = red_qxl_disconnect_cursor_peer;
     channel_class->migrate = red_qxl_cursor_migrate;
 }
diff --git a/server/display-channel.c b/server/display-channel.c
index ab36dea8..407d5a0f 100644
--- a/server/display-channel.c
+++ b/server/display-channel.c
@@ -2263,6 +2263,7 @@ DisplayChannel* display_channel_new(RedsState *reds,
                            "n-surfaces", n_surfaces,
                            "video-codecs", video_codecs,
                            "handle-acks", TRUE,
+                           "thread-dispatcher", red_qxl_get_dispatcher(qxl),
                            NULL);
     if (display) {
         display_channel_set_stream_video(display, stream_video);
@@ -2295,66 +2296,55 @@ static void red_qxl_set_display_peer(RedChannel *channel, RedClient *client,
                                      RedsStream *stream, int migration,
                                      RedChannelCapabilities *caps)
 {
-    RedWorkerMessageDisplayConnect payload = {0,};
-    QXLInstance *qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    Dispatcher *dispatcher = red_qxl_get_dispatcher(qxl);
+    DisplayChannelClient *dcc;
 
-    spice_debug("%s", "");
-    // get a reference potentially the main channel can be destroyed in
-    // the main thread causing RedClient to be destroyed before using it
-    payload.client = g_object_ref(client);
-    payload.stream = stream;
-    payload.migration = migration;
-    red_channel_capabilities_init(&payload.caps, caps);
+    spice_debug("connect new client");
+    spice_return_if_fail(channel);
 
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_DISPLAY_CONNECT,
-                            &payload);
+    dcc = dcc_new(DISPLAY_CHANNEL(channel), client, stream, migration, caps,
+                  //worker->image_compression, worker->jpeg_state, worker->zlib_glz_state);
+                  0, 0, 0);
+    if (!dcc) {
+        return;
+    }
+    display_channel_update_compression(DISPLAY_CHANNEL(channel), dcc);
+    //guest_set_client_capabilities(worker);
+    dcc_start(dcc);
 }
 
 static void red_qxl_disconnect_display_peer(RedChannel *channel, RedChannelClient *rcc)
 {
-    RedWorkerMessageDisplayDisconnect payload;
-    QXLInstance *qxl;
-    Dispatcher *dispatcher;
-
-    if (!channel) {
-        return;
-    }
-
-    qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    dispatcher = red_qxl_get_dispatcher(qxl);
+    spice_debug("disconnect display client");
+    spice_assert(rcc);
 
-    spice_printerr("");
-    payload.rcc = rcc;
+    //guest_set_client_capabilities(worker);
 
-    // TODO: we turned it to be sync, due to client_destroy . Should we support async? - for this we will need ref count
-    // for channels
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_DISPLAY_DISCONNECT,
-                            &payload);
+    red_channel_client_disconnect(rcc);
 }
 
-static void red_qxl_display_migrate(RedChannel *channel, RedChannelClient *rcc)
+static void red_migrate_display(DisplayChannel *display, RedChannelClient *rcc)
 {
-    RedWorkerMessageDisplayMigrate payload;
-    QXLInstance *qxl;
-    Dispatcher *dispatcher;
-    uint32_t type, id;
-
-    if (!channel) {
-        return;
+    /* We need to stop the streams, and to send upgrade_items to the client.
+     * Otherwise, (1) the client might display lossy regions that we don't track
+     * (streams are not part of the migration data) (2) streams_timeout may occur
+     * after the MIGRATE message has been sent. This can result in messages
+     * being sent to the client after MSG_MIGRATE and before MSG_MIGRATE_DATA (e.g.,
+     * STREAM_CLIP, STREAM_DESTROY, DRAW_COPY)
+     * No message besides MSG_MIGRATE_DATA should be sent after MSG_MIGRATE.
+     * Notice that detach_and_stop_streams won't lead to any dev ram changes, since
+     * handle_dev_stop already took care of releasing all the dev ram resources.
+     */
+    stream_detach_and_stop(display);
+    if (red_channel_client_is_connected(rcc)) {
+        red_channel_client_default_migrate(rcc);
     }
-    g_object_get(channel, "channel-type", &type, "id", &id,
-                 NULL);
+}
 
-    qxl = common_graphics_channel_get_qxl(COMMON_GRAPHICS_CHANNEL(channel));
-    dispatcher = red_qxl_get_dispatcher(qxl);
-    spice_printerr("channel type %u id %u", type, id);
-    payload.rcc = rcc;
-    dispatcher_send_message(dispatcher,
-                            RED_WORKER_MESSAGE_DISPLAY_MIGRATE,
-                            &payload);
+static void red_qxl_display_migrate(RedChannel *channel, RedChannelClient *rcc)
+{
+    spice_debug("migrate display client");
+    spice_assert(rcc);
+    red_migrate_display(DISPLAY_CHANNEL(channel), rcc);
 }
 
 static void
diff --git a/server/red-channel.c b/server/red-channel.c
index b8e2e886..a24c9a49 100644
--- a/server/red-channel.c
+++ b/server/red-channel.c
@@ -26,6 +26,7 @@
 
 #include "red-channel.h"
 #include "red-channel-client.h"
+#include "red-qxl.h"
 #include "reds.h"
 #include "reds-stream.h"
 #include "main-dispatcher.h"
@@ -96,6 +97,8 @@ struct RedChannelPrivate
     pthread_t thread_id;
     RedsState *reds;
     RedStatNode stat;
+
+    void *dispatcher;
 };
 
 enum {
@@ -467,15 +470,6 @@ void red_channel_disconnect(RedChannel *channel)
     g_list_foreach(channel->priv->clients, (GFunc)red_channel_client_disconnect, NULL);
 }
 
-void red_channel_connect(RedChannel *channel, RedClient *client,
-                         RedsStream *stream, int migration,
-                         RedChannelCapabilities *caps)
-{
-    RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
-    g_return_if_fail(klass->connect);
-    klass->connect(channel, client, stream, migration, caps);
-}
-
 void red_channel_apply_clients(RedChannel *channel, channel_client_callback cb)
 {
     g_list_foreach(channel->priv->clients, (GFunc)cb, NULL);
@@ -706,11 +700,58 @@ void red_channel_migrate_client(RedChannel *channel, RedChannelClient *rcc)
     RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
     g_return_if_fail(klass->migrate);
     klass->migrate(channel, rcc);
+
+    if (channel->priv->dispatcher) {
+        RedWorkerMessageChannelMigrateClient payload;
+
+        payload.rcc = rcc;
+
+        dispatcher_send_message(channel->priv->dispatcher,
+                                RED_WORKER_MESSAGE_CURSOR_MIGRATE,
+                                &payload);
+    } else {
+        klass->migrate(channel, rcc);
+    }
+}
+
+void red_channel_connect(RedChannel *channel, RedClient *client,
+                         RedsStream *stream, int migration,
+                         RedChannelCapabilities *caps)
+{
+    RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
+    g_return_if_fail(klass->connect);
+
+    if (channel->priv->dispatcher) {
+        RedWorkerMessageChannelConnectClient payload;
+
+        payload.channel = channel;
+	payload.client = g_object_ref(client);
+	payload.stream = stream;
+	payload.migration = migration;
+	red_channel_capabilities_init(&payload.caps, caps);
+
+        dispatcher_send_message(channel->priv->dispatcher,
+                                RED_WORKER_MESSAGE_CHANNEL_CONNECT_CLIENT,
+                                &payload);
+    } else {
+	klass->connect(channel, client, stream, migration, caps);
+    }
 }
 
 void red_channel_disconnect_client(RedChannel *channel, RedChannelClient *rcc)
 {
     RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
     g_return_if_fail(klass->disconnect);
-    klass->disconnect(channel, rcc);
+
+    if (channel->priv->dispatcher) {
+        RedWorkerMessageChannelDisconnectClient payload;
+
+        payload.rcc = rcc;
+
+        dispatcher_send_message(channel->priv->dispatcher,
+                                RED_WORKER_MESSAGE_CHANNEL_DISCONNECT_CLIENT,
+                                &payload);
+    } else {
+        klass->disconnect(channel, rcc);
+    }
 }
diff --git a/server/red-qxl.h b/server/red-qxl.h
index 93cc4d0f..60bd410f 100644
--- a/server/red-qxl.h
+++ b/server/red-qxl.h
@@ -117,9 +117,14 @@ enum {
     RED_WORKER_MESSAGE_GL_DRAW_ASYNC,
     RED_WORKER_MESSAGE_SET_VIDEO_CODECS,
 
+    RED_WORKER_MESSAGE_CHANNEL_CONNECT_CLIENT,
+    RED_WORKER_MESSAGE_CHANNEL_DISCONNECT_CLIENT,
+    RED_WORKER_MESSAGE_CHANNEL_MIGRATE_CLIENT,
+
     /* close worker thread */
     RED_WORKER_MESSAGE_CLOSE_WORKER,
 
+
     RED_WORKER_MESSAGE_COUNT // LAST
 };
 
@@ -153,6 +158,22 @@ typedef struct RedWorkerMessageCursorMigrate {
     RedChannelClient *rcc;
 } RedWorkerMessageCursorMigrate;
 
+typedef struct RedWorkerMessageChannelConnectClient {
+    RedChannel *channel;
+    RedClient *client;
+    RedsStream *stream;
+    int migration;
+    RedChannelCapabilities caps;   // red_worker should reset
+} RedWorkerMessageChannelConnectClient;
+
+typedef struct RedWorkerMessageChannelDisconnectClient {
+    RedChannelClient *rcc;
+} RedWorkerMessageChannelDisconnectClient;
+
+typedef struct RedWorkerMessageChannelMigrateClient {
+    RedChannelClient *rcc;
+} RedWorkerMessageChannelMigrateClient;
+
 typedef struct RedWorkerMessageUpdate {
     uint32_t surface_id;
     QXLRect * qxl_area;
diff --git a/server/red-worker.c b/server/red-worker.c
index c62c0486..82fd5290 100644
--- a/server/red-worker.c
+++ b/server/red-worker.c
@@ -845,6 +845,45 @@ static void handle_dev_cursor_migrate(void *opaque, void *payload)
     cursor_channel_client_migrate(rcc);
 }
 
+static void handle_channel_connect_client(void *opaque, void *payload)
+{
+    RedWorkerMessageChannelConnectClient *msg = payload;
+    RedChannel *channel = msg->channel;
+    g_return_if_fail(channel);
+    RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
+
+    g_debug("connect channel client");
+    g_return_if_fail(klass->connect);
+    klass->connect(channel, msg->client, msg->stream, msg->migration, &msg->caps);
+    g_object_unref(msg->client);
+    red_channel_capabilities_reset(&msg->caps);
+}
+
+static void handle_channel_disconnect_client(void *opaque, void *payload)
+{
+    RedWorkerMessageChannelDisconnectClient *msg = payload;
+    RedChannelClient *rcc = msg->rcc;
+    g_return_if_fail(rcc);
+    RedChannel *channel = red_channel_client_get_channel(rcc);
+    RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
+
+    g_debug("disconnect channel client");
+    g_return_if_fail(klass->disconnect);
+    klass->disconnect(channel, rcc);
+}
+
+static void handle_channel_migrate_client(void *opaque, void *payload)
+{
+    RedWorkerMessageChannelMigrateClient *msg = payload;
+    RedChannelClient *rcc = msg->rcc;
+    g_return_if_fail(rcc);
+    RedChannel *channel = red_channel_client_get_channel(rcc);
+    RedChannelClass *klass = RED_CHANNEL_GET_CLASS(channel);
+
+    g_debug("migrate channel client");
+    klass->migrate(channel, rcc);
+}
+
 static void handle_dev_set_compression(void *opaque, void *payload)
 {
     RedWorkerMessageSetCompression *msg = payload;
@@ -1063,6 +1102,21 @@ static void register_callbacks(Dispatcher *dispatcher)
                                 sizeof(RedWorkerMessageCursorMigrate),
                                 DISPATCHER_NONE);
     dispatcher_register_handler(dispatcher,
+                                RED_WORKER_MESSAGE_CHANNEL_CONNECT_CLIENT,
+                                handle_channel_connect_client,
+                                sizeof(RedWorkerMessageChannelConnectClient),
+                                DISPATCHER_NONE);
+    dispatcher_register_handler(dispatcher,
+                                RED_WORKER_MESSAGE_CHANNEL_DISCONNECT_CLIENT,
+                                handle_channel_disconnect_client,
+                                sizeof(RedWorkerMessageChannelDisconnectClient),
+                                DISPATCHER_ACK);
+    dispatcher_register_handler(dispatcher,
+                                RED_WORKER_MESSAGE_CHANNEL_MIGRATE_CLIENT,
+                                handle_channel_migrate_client,
+                                sizeof(RedWorkerMessageChannelMigrateClient),
+                                DISPATCHER_NONE);
+    dispatcher_register_handler(dispatcher,
                                 RED_WORKER_MESSAGE_UPDATE,
                                 handle_dev_update,
                                 sizeof(RedWorkerMessageUpdate),