[Spice-devel] [spice-gtk PATCH v4 1/6] giopipe: don't fail on create_source

[Victor Toso]

PipeInputStream and PipeOutputStream should not fail when creating
GPollableStream source as this currently does not work with default
write_all and read_all functions;

This patch removes the g_return_val_if_fail but keeps a g_debug in order
to track problems;

In order to avoid creating zombie GSource in create_source of both
PipeInputStream and PipeOutputStream, we track all created GSources and
set them to be dispatched when data is available to read/write. It is
worth to mention that concurrent write/read is not possible with current
giopipe and only the last created GSource will read the data as it is
dispatched first.
---
 gtk/giopipe.c | 60 +++++++++++++++++++++++++++++++++++++----------------------
 1 file changed, 38 insertions(+), 22 deletions(-)

diff --git a/gtk/giopipe.c b/gtk/giopipe.c
index 50edb5b..cdfa070 100644
--- a/gtk/giopipe.c
+++ b/gtk/giopipe.c
@@ -45,6 +45,7 @@ struct _PipeInputStream
      */
     gboolean peer_closed;
     GSource *source;
+    GList *created_sources;
 };
 
 struct _PipeInputStreamClass
@@ -70,6 +71,7 @@ struct _PipeOutputStream
     gsize count;
     gboolean peer_closed;
     GSource *source;
+    GList *created_sources;
 };
 
 struct _PipeOutputStreamClass
@@ -121,11 +123,26 @@ pipe_input_stream_read (GInputStream  *stream,
 }
 
 static void
+set_all_sources_ready (GList *sources)
+{
+    GList *it;
+    for (it = sources; it != NULL; it = it->next) {
+        GSource *s = it->data;
+        if (s != NULL && !g_source_is_destroyed(s))
+            g_source_set_ready_time(s, 0);
+    }
+    g_list_free_full (sources, (GDestroyNotify) g_source_unref);
+}
+
+static void
 pipe_input_stream_check_source (PipeInputStream *self)
 {
     if (self->source && !g_source_is_destroyed(self->source) &&
-        g_pollable_input_stream_is_readable(G_POLLABLE_INPUT_STREAM(self)))
-        g_source_set_ready_time(self->source, 0);
+        g_pollable_input_stream_is_readable(G_POLLABLE_INPUT_STREAM(self))) {
+        set_all_sources_ready(self->created_sources);
+        self->created_sources = NULL;
+        self->source = NULL;
+    }
 }
 
 static gboolean
@@ -193,10 +210,9 @@ pipe_input_stream_dispose(GObject *object)
         self->peer = NULL;
     }
 
-    if (self->source) {
-        g_source_unref(self->source);
-        self->source = NULL;
-    }
+    g_list_free_full (self->created_sources, (GDestroyNotify) g_source_unref);
+    self->created_sources = NULL;
+    self->source = NULL;
 
     G_OBJECT_CLASS(pipe_input_stream_parent_class)->dispose (object);
 }
@@ -234,14 +250,13 @@ pipe_input_stream_create_source (GPollableInputStream *stream,
     PipeInputStream *self = PIPE_INPUT_STREAM(stream);
     GSource *pollable_source;
 
-    g_return_val_if_fail (self->source == NULL ||
-                          g_source_is_destroyed (self->source), NULL);
-
-    if (self->source && g_source_is_destroyed (self->source))
-        g_source_unref (self->source);
+    if (self->source != NULL && !g_source_is_destroyed (self->source))
+        g_debug ("%s: GPollableSource already exists %p - This could lead to data loss (%ld)",
+                 G_STRFUNC, self->source, self->read);
 
     pollable_source = g_pollable_source_new_full (self, NULL, cancellable);
     self->source = g_source_ref (pollable_source);
+    self->created_sources = g_list_prepend (self->created_sources, self->source);
 
     return pollable_source;
 }
@@ -319,10 +334,9 @@ pipe_output_stream_dispose(GObject *object)
         self->peer = NULL;
     }
 
-    if (self->source) {
-        g_source_unref(self->source);
-        self->source = NULL;
-    }
+    g_list_free_full (self->created_sources, (GDestroyNotify) g_source_unref);
+    self->created_sources = NULL;
+    self->source = NULL;
 
     G_OBJECT_CLASS(pipe_output_stream_parent_class)->dispose (object);
 }
@@ -331,8 +345,11 @@ static void
 pipe_output_stream_check_source (PipeOutputStream *self)
 {
     if (self->source && !g_source_is_destroyed(self->source) &&
-        g_pollable_output_stream_is_writable(G_POLLABLE_OUTPUT_STREAM(self)))
-        g_source_set_ready_time(self->source, 0);
+        g_pollable_output_stream_is_writable(G_POLLABLE_OUTPUT_STREAM(self))) {
+        set_all_sources_ready(self->created_sources);
+        self->created_sources = NULL;
+        self->source = NULL;
+    }
 }
 
 static gboolean
@@ -416,14 +433,13 @@ pipe_output_stream_create_source (GPollableOutputStream *stream,
     PipeOutputStream *self = PIPE_OUTPUT_STREAM(stream);
     GSource *pollable_source;
 
-    g_return_val_if_fail (self->source == NULL ||
-                          g_source_is_destroyed (self->source), NULL);
-
-    if (self->source && g_source_is_destroyed (self->source))
-        g_source_unref (self->source);
+    if (self->source != NULL && !g_source_is_destroyed (self->source))
+        g_debug ("%s: GPollableSource already exists %p - This could lead to data loss (%ld)",
+                 G_STRFUNC, self->source, self->count);
 
     pollable_source = g_pollable_source_new_full (self, NULL, cancellable);
     self->source = g_source_ref (pollable_source);
+    self->created_sources = g_list_prepend (self->created_sources, self->source);
 
     return pollable_source;
 }
-- 
2.4.2