server/red_channel.c inroducing client_cbs

client_cbs are supposed to be called from client context (reds). This patch will be used
in future patches for relacing reds::Channel with RedChannel in order to eliminate redundancy.

server/red_channel.c

@@ -417,6 +417,24 @@ error:
     return NULL;
 }
 
+static void red_channel_client_default_connect(RedChannel *channel, RedClient *client,
+                                               RedsStream *stream,
+                                               int migration,
+                                               int num_common_caps, uint32_t *common_caps,
+                                               int num_caps, uint32_t *caps)
+{
+    red_error("not implemented");
+}
+
+static void red_channel_client_default_disconnect(RedChannelClient *base)
+{
+    red_channel_client_disconnect(base);
+}
+
+static void red_channel_client_default_migrate(RedChannelClient *base)
+{
+}
+
 RedChannel *red_channel_create(int size,
                                SpiceCoreInterface *core,
                                int migrate, int handle_acks,
@@ -424,6 +442,7 @@ RedChannel *red_channel_create(int size,
                                ChannelCbs *channel_cbs)
 {
     RedChannel *channel;
+    ClientCbs client_cbs;
 
     ASSERT(size >= sizeof(*channel));
     ASSERT(channel_cbs->config_socket && channel_cbs->disconnect && handle_message &&
@@ -457,6 +476,14 @@ RedChannel *red_channel_create(int size,
     channel->outgoing_cb.on_msg_done = red_channel_peer_on_out_msg_done;
     channel->outgoing_cb.on_output = red_channel_client_on_output;
 
+    client_cbs.connect = red_channel_client_default_connect;
+    client_cbs.disconnect = red_channel_client_default_disconnect;
+    client_cbs.migrate = red_channel_client_default_migrate;
+
+    red_channel_register_client_cbs(channel, &client_cbs);
+
+    channel->thread_id = pthread_self();
+
     channel->shut = 0; // came here from inputs, perhaps can be removed? XXX
     channel->out_bytes_counter = 0;
     return channel;
@@ -492,6 +519,20 @@ RedChannel *red_channel_create_parser(int size,
     return channel;
 }
 
+void red_channel_register_client_cbs(RedChannel *channel, ClientCbs *client_cbs)
+{
+    ASSERT(client_cbs->connect);
+    channel->client_cbs.connect = client_cbs->connect;
+
+    if (client_cbs->disconnect) {
+        channel->client_cbs.disconnect = client_cbs->disconnect;
+    }
+
+    if (client_cbs->migrate) {
+        channel->client_cbs.migrate = client_cbs->migrate;
+    }
+}
+
 void red_channel_client_destroy(RedChannelClient *rcc)
 {
     red_channel_client_disconnect(rcc);
@@ -1102,6 +1143,8 @@ RedClient *red_client_new()
 
     client = spice_malloc0(sizeof(RedClient));
     ring_init(&client->channels);
+    client->thread_id = pthread_self();
+
     return client;
 }
 
@@ -1121,11 +1164,17 @@ void red_client_destroy(RedClient *client)
     RedChannelClient *rcc;
 
     red_printf("destroy client with #channels %d", client->channels_num);
+    ASSERT(pthread_equal(pthread_self(), client->thread_id));
     RING_FOREACH_SAFE(link, next, &client->channels) {
         // some channels may be in other threads, so disconnection
         // is not synchronous.
         rcc = SPICE_CONTAINEROF(link, RedChannelClient, client_link);
-        rcc->channel->channel_cbs.disconnect(rcc); // this may call another thread. it also frees. (eventually - doesn't have to be in sync)
+        // some channels may be in other threads. However we currently
+        // assume disconnect is synchronous (we changed the dispatcher
+        // to wait for disconnection)
+        // TODO: should we go back to async. For this we need to use
+        // ref count for channel clients.
+        rcc->channel->client_cbs.disconnect(rcc);
     }
     free(client);
 }
@@ -1140,7 +1189,7 @@ void red_client_disconnect(RedClient *client)
         // some channels may be in other threads, so disconnection
         // is not synchronous.
         rcc = SPICE_CONTAINEROF(link, RedChannelClient, client_link);
-        rcc->channel->channel_cbs.disconnect(rcc);
+        rcc->channel->client_cbs.disconnect(rcc);
     }
 }
 

server/red_channel.h

@@ -23,6 +23,7 @@
 #define _H_RED_CHANNEL
 
 #include "red_common.h"
+#include <pthread.h>
 #include "reds.h"
 #include "spice.h"
 #include "ring.h"
@@ -138,6 +139,15 @@ typedef uint64_t (*channel_handle_migrate_data_proc)(RedChannelClient *base,
 typedef uint64_t (*channel_handle_migrate_data_get_serial_proc)(RedChannelClient *base,
                                             uint32_t size, void *message);
 
+
+typedef void (*channel_client_connect_proc)(RedChannel *channel, RedClient *client, RedsStream *stream,
+                                            int migration, int num_common_caps, uint32_t *common_caps,
+                                            int num_caps, uint32_t *caps);
+typedef void (*channel_client_disconnect_proc)(RedChannelClient *base);
+typedef void (*channel_client_migrate_proc)(RedChannelClient *base);
+
+// TODO: add ASSERTS for thread_id  in client and channel calls
+//
 /*
  * callbacks that are triggered from channel client stream events.
  * They are called from the thread that listen to the stream events.
@@ -155,6 +165,17 @@ typedef struct {
     channel_handle_migrate_data_get_serial_proc handle_migrate_data_get_serial;
 } ChannelCbs;
 
+
+/*
+ * callbacks that are triggered from client events.
+ * They should be called from the thread that handles the RedClient
+ */
+typedef struct {
+    channel_client_connect_proc connect;
+    channel_client_disconnect_proc disconnect;
+    channel_client_migrate_proc migrate;
+} ClientCbs;
+
 struct RedChannelClient {
     RingItem channel_link;
     RingItem client_link;
@@ -197,6 +218,7 @@ struct RedChannel {
     IncomingHandlerInterface incoming_cb;
 
     ChannelCbs channel_cbs;
+    ClientCbs client_cbs;
 
     /* Stuff below added for Main and Inputs channels switch to RedChannel
      * (might be removed later) */
@@ -204,6 +226,8 @@ struct RedChannel {
     channel_on_outgoing_error_proc on_outgoing_error;
     int shut; /* signal channel is to be closed */
 
+    // TODO: when different channel_clients are in different threads from Channel -> need to protect!
+    pthread_t thread_id;
 #ifdef RED_STATISTICS
     uint64_t *out_bytes_counter;
 #endif
@@ -228,6 +252,8 @@ RedChannel *red_channel_create_parser(int size,
                                channel_on_outgoing_error_proc outgoing_error,
                                ChannelCbs *channel_cbs);
 
+void red_channel_register_client_cbs(RedChannel *channel, ClientCbs *client_cbs);
+
 RedChannelClient *red_channel_client_create(int size, RedChannel *channel, RedClient *client,
                                             RedsStream *stream);
 
@@ -379,6 +405,8 @@ struct RedClient {
     int channels_num;
     int disconnecting;
     MainChannelClient *mcc;
+
+    pthread_t thread_id;
 };
 
 RedClient *red_client_new();