[Spice-devel] [PATCH spice-gtk 01/10] spice-channel: Allow spice_msg_out_send to be called from multiple threads

[Hans de Goede]

This is a preparation patch for handling usb packet completion in a
separate thread.

Signed-off-by: Hans de Goede <hdegoede at redhat.com>
---
 gtk/spice-channel-priv.h |    3 +++
 gtk/spice-channel.c      |   38 ++++++++++++++++++++++++++++++++++----
 2 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/gtk/spice-channel-priv.h b/gtk/spice-channel-priv.h
index ed5f3a7..b9e81f8 100644
--- a/gtk/spice-channel-priv.h
+++ b/gtk/spice-channel-priv.h
@@ -95,6 +95,9 @@ struct _SpiceChannelPrivate {
 
     struct wait_queue           wait;
     GQueue                      xmit_queue;
+    gboolean                    xmit_queue_blocked;
+    GStaticMutex                xmit_queue_lock;
+    GThread                     *main_thread;
 
     char                        name[16];
     enum spice_channel_state    state;
diff --git a/gtk/spice-channel.c b/gtk/spice-channel.c
index b6779ba..873f9b3 100644
--- a/gtk/spice-channel.c
+++ b/gtk/spice-channel.c
@@ -105,6 +105,8 @@ static void spice_channel_init(SpiceChannel *channel)
     c->remote_common_caps = g_array_new(FALSE, TRUE, sizeof(guint32));
     spice_channel_set_common_capability(channel, SPICE_COMMON_CAP_PROTOCOL_AUTH_SELECTION);
     g_queue_init(&c->xmit_queue);
+    g_static_mutex_init(&c->xmit_queue_lock);
+    c->main_thread = g_thread_self();
 }
 
 static void spice_channel_constructed(GObject *gobject)
@@ -535,16 +537,35 @@ void spice_msg_out_unref(SpiceMsgOut *out)
 }
 
 /* system context */
+static gboolean spice_channel_idle_wakeup(gpointer user_data)
+{
+    SpiceChannel *channel = SPICE_CHANNEL(user_data);
+
+    spice_channel_wakeup(channel);
+    g_object_unref(channel);
+
+    return FALSE;
+}
+
+/* system context */
 G_GNUC_INTERNAL
 void spice_msg_out_send(SpiceMsgOut *out)
 {
     g_return_if_fail(out != NULL);
     g_return_if_fail(out->channel != NULL);
 
-    g_queue_push_tail(&out->channel->priv->xmit_queue, out);
+    g_static_mutex_lock(&out->channel->priv->xmit_queue_lock);
+    if (!out->channel->priv->xmit_queue_blocked)
+        g_queue_push_tail(&out->channel->priv->xmit_queue, out);
+    g_static_mutex_unlock(&out->channel->priv->xmit_queue_lock);
 
     /* TODO: we currently flush/wakeup immediately all buffered messages */
-    spice_channel_wakeup(out->channel);
+    if (g_thread_self() != out->channel->priv->main_thread)
+        /* We use g_timeout_add_full so that can specify the priority */
+        g_timeout_add_full(G_PRIORITY_HIGH, 0, spice_channel_idle_wakeup,
+                           g_object_ref(out->channel), NULL);
+    else
+        spice_channel_wakeup(out->channel);
 }
 
 /* coroutine context */
@@ -1784,8 +1805,13 @@ static void spice_channel_iterate_write(SpiceChannel *channel)
     SpiceChannelPrivate *c = channel->priv;
     SpiceMsgOut *out;
 
-    while ((out = g_queue_pop_head(&c->xmit_queue)))
-        spice_channel_write_msg(channel, out);
+    do {
+        g_static_mutex_lock(&c->xmit_queue_lock);
+        out = g_queue_pop_head(&c->xmit_queue);
+        g_static_mutex_unlock(&c->xmit_queue_lock);
+        if (out)
+            spice_channel_write_msg(channel, out);
+    } while (out);
 }
 
 /* coroutine context */
@@ -2065,6 +2091,7 @@ static gboolean channel_connect(SpiceChannel *channel)
         }
     }
     c->state = SPICE_CHANNEL_STATE_CONNECTING;
+    c->xmit_queue_blocked = FALSE;
 
     g_return_val_if_fail(c->sock == NULL, FALSE);
     g_object_ref(G_OBJECT(channel)); /* Unref'd when co-routine exits */
@@ -2165,8 +2192,11 @@ static void channel_disconnect(SpiceChannel *channel)
     c->peer_msg = NULL;
     c->peer_pos = 0;
 
+    g_static_mutex_lock(&c->xmit_queue_lock);
+    c->xmit_queue_blocked = TRUE; /* Disallow queuing new messages */
     g_queue_foreach(&c->xmit_queue, (GFunc)spice_msg_out_unref, NULL);
     g_queue_clear(&c->xmit_queue);
+    g_static_mutex_unlock(&c->xmit_queue_lock);
 
     g_array_set_size(c->remote_common_caps, 0);
     g_array_set_size(c->remote_caps, 0);
-- 
1.7.7.4