soup-cache-input-stream: fixed a reference cycle
authorSergio Villar Senin <svillar@igalia.com>
Fri, 15 Feb 2013 20:40:28 +0000 (21:40 +0100)
committerSergio Villar Senin <svillar@igalia.com>
Mon, 18 Feb 2013 10:03:15 +0000 (11:03 +0100)
Fixed a reference cycle between SoupCacheInputStream and a
GTask. This reference cycle was preventing the input stream from
being freed when the SoupMessage was cancelled before the stream
was completely read. Furthermore the SoupCache was not able to
complete the pending request because the SoupCacheInputStream was
not aware of the message cancellation and remained waiting forever.

In order to remove the cycle
soup_cache_input_stream_cache(_finish) were removed and the
caching operation finalization is now reported via a signal.
SoupCacheInputStream also implements close_fn which properly
finalizes any pending caching operation.

https://bugzilla.gnome.org/show_bug.cgi?id=682527

libsoup/soup-cache-input-stream.c
libsoup/soup-cache-input-stream.h
libsoup/soup-cache.c

index a44652a..7815cec 100644 (file)
@@ -9,6 +9,7 @@
 
 #include <glib/gi18n-lib.h>
 #include "soup-cache-input-stream.h"
+#include "soup-marshal.h"
 #include "soup-message-body.h"
 
 static void soup_cache_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
@@ -26,16 +27,23 @@ enum {
        LAST_PROP
 };
 
+enum {
+       CACHING_FINISHED,
+
+       LAST_SIGNAL
+};
+
+static guint signals[LAST_SIGNAL] = { 0 };
+
 struct _SoupCacheInputStreamPrivate
 {
        GOutputStream *output_stream;
+       GCancellable *cancellable;
        gsize bytes_written;
 
        gboolean read_finished;
        SoupBuffer *current_writing_buffer;
        GQueue *buffer_queue;
-
-       GTask *task;
 };
 
 static void soup_cache_input_stream_write_next_buffer (SoupCacheInputStream *istream);
@@ -45,21 +53,10 @@ notify_and_clear (SoupCacheInputStream *istream, GError *error)
 {
        SoupCacheInputStreamPrivate *priv = istream->priv;
 
-       if (error)
-               g_task_return_error (priv->task, error);
-       else
-               g_task_return_int (priv->task, priv->bytes_written);
+       g_signal_emit (istream, signals[CACHING_FINISHED], 0, priv->bytes_written, error);
 
+       g_clear_object (&priv->cancellable);
        g_clear_object (&priv->output_stream);
-       g_clear_object (&priv->task);
-}
-
-gsize
-soup_cache_input_stream_cache_finish (SoupCacheInputStream  *istream,
-                                     GAsyncResult          *result,
-                                     GError               **error)
-{
-       return g_task_propagate_int (G_TASK (result), error);
 }
 
 static inline void
@@ -91,25 +88,11 @@ file_replaced_cb (GObject      *source,
        priv->output_stream = (GOutputStream *) g_file_replace_finish (G_FILE (source), res, &error);
 
        if (error)
-               g_task_return_error (priv->task, error);
+               notify_and_clear (istream, error);
        else
                try_write_next_buffer (istream);
-}
-
-void
-soup_cache_input_stream_cache (SoupCacheInputStream  *istream,
-                              GFile                 *file,
-                              GCancellable          *cancellable,
-                              GAsyncReadyCallback    callback,
-                              gpointer               user_data)
-{
-       SoupCacheInputStreamPrivate *priv = istream->priv;
 
-       priv->task = g_task_new (istream, cancellable, callback, user_data);
-
-       g_file_replace_async (file, NULL, FALSE,
-                             G_FILE_CREATE_PRIVATE | G_FILE_CREATE_REPLACE_DESTINATION,
-                             G_PRIORITY_LOW, cancellable, file_replaced_cb, istream);
+       g_object_unref (istream);
 }
 
 static void
@@ -158,22 +141,13 @@ soup_cache_input_stream_set_property (GObject *object,
 }
 
 static void
-soup_cache_input_stream_dispose (GObject *object)
-{
-       SoupCacheInputStreamPrivate *priv = SOUP_CACHE_INPUT_STREAM (object)->priv;
-
-       g_clear_object (&priv->output_stream);
-       g_clear_object (&priv->task);
-
-       G_OBJECT_CLASS (soup_cache_input_stream_parent_class)->dispose (object);
-}
-
-static void
 soup_cache_input_stream_finalize (GObject *object)
 {
        SoupCacheInputStream *self = (SoupCacheInputStream *)object;
        SoupCacheInputStreamPrivate *priv = self->priv;
 
+       g_clear_object (&priv->cancellable);
+       g_clear_object (&priv->output_stream);
        g_clear_pointer (&priv->current_writing_buffer, soup_buffer_free);
        g_queue_free_full (priv->buffer_queue, (GDestroyNotify) soup_buffer_free);
 
@@ -219,7 +193,6 @@ soup_cache_input_stream_write_next_buffer (SoupCacheInputStream *istream)
        int priority;
 
        g_assert (priv->output_stream && !g_output_stream_is_closed (priv->output_stream));
-       g_assert (priv->task);
 
        g_clear_pointer (&priv->current_writing_buffer, soup_buffer_free);
        priv->current_writing_buffer = buffer;
@@ -230,7 +203,7 @@ soup_cache_input_stream_write_next_buffer (SoupCacheInputStream *istream)
                priority = G_PRIORITY_LOW;
 
        g_output_stream_write_async (priv->output_stream, buffer->data, buffer->length,
-                                    priority, g_task_get_cancellable (priv->task),
+                                    priority, priv->cancellable,
                                     (GAsyncReadyCallback) write_ready_cb,
                                     g_object_ref (istream));
 }
@@ -252,7 +225,7 @@ read_internal (GInputStream  *stream,
        nread = g_pollable_stream_read (base_stream, buffer, count, blocking,
                                        cancellable, error);
 
-       if (G_UNLIKELY (nread == -1 || priv->read_finished || !priv->task))
+       if (G_UNLIKELY (nread == -1 || priv->read_finished))
                return nread;
 
        if (nread == 0) {
@@ -299,6 +272,33 @@ soup_cache_input_stream_pollable_init (GPollableInputStreamInterface *pollable_i
        pollable_interface->read_nonblocking = soup_cache_input_stream_read_nonblocking;
 }
 
+static gboolean
+soup_cache_input_stream_close_fn (GInputStream  *stream,
+                                 GCancellable  *cancellable,
+                                 GError       **error)
+{
+       SoupCacheInputStream *istream = SOUP_CACHE_INPUT_STREAM (stream);
+       SoupCacheInputStreamPrivate *priv = istream->priv;
+
+       if (!priv->read_finished) {
+               if (priv->output_stream) {
+                       /* Cancel any pending write operation or return an error if none. */
+                       if (g_output_stream_has_pending (priv->output_stream))
+                               g_cancellable_cancel (priv->cancellable);
+                       else {
+                               GError *error = NULL;
+                               g_set_error_literal (&error, G_IO_ERROR, G_IO_ERROR_PARTIAL_INPUT,
+                                                    _("Failed to completely cache the resource"));
+                               notify_and_clear (istream, error);
+                       }
+               } else if (priv->cancellable)
+                       /* The file_replace_async() hasn't finished yet */
+                       g_cancellable_cancel (priv->cancellable);
+       }
+
+       return G_INPUT_STREAM_CLASS (soup_cache_input_stream_parent_class)->close_fn (stream, cancellable, error);
+}
+
 static void
 soup_cache_input_stream_class_init (SoupCacheInputStreamClass *klass)
 {
@@ -309,10 +309,10 @@ soup_cache_input_stream_class_init (SoupCacheInputStreamClass *klass)
 
        gobject_class->get_property = soup_cache_input_stream_get_property;
        gobject_class->set_property = soup_cache_input_stream_set_property;
-       gobject_class->dispose = soup_cache_input_stream_dispose;
        gobject_class->finalize = soup_cache_input_stream_finalize;
 
        istream_class->read_fn = soup_cache_input_stream_read_fn;
+       istream_class->close_fn = soup_cache_input_stream_close_fn;
 
        g_object_class_install_property (gobject_class, PROP_OUTPUT_STREAM,
                                         g_param_spec_object ("output-stream", "Output stream",
@@ -321,13 +321,32 @@ soup_cache_input_stream_class_init (SoupCacheInputStreamClass *klass)
                                                              G_PARAM_READWRITE |
                                                              G_PARAM_CONSTRUCT_ONLY |
                                                              G_PARAM_STATIC_STRINGS));
+
+       signals[CACHING_FINISHED] =
+               g_signal_new ("caching-finished",
+                             G_OBJECT_CLASS_TYPE (gobject_class),
+                             G_SIGNAL_RUN_FIRST,
+                             G_STRUCT_OFFSET (SoupCacheInputStreamClass, caching_finished),
+                             NULL, NULL,
+                             _soup_marshal_NONE__INT_BOXED,
+                             G_TYPE_NONE, 2,
+                             G_TYPE_INT, G_TYPE_ERROR);
 }
 
 GInputStream *
-soup_cache_input_stream_new (GInputStream *base_stream)
+soup_cache_input_stream_new (GInputStream *base_stream,
+                            GFile        *file)
 {
-       return g_object_new (SOUP_TYPE_CACHE_INPUT_STREAM,
-                            "base-stream", base_stream,
-                            "close-base-stream", FALSE,
-                            NULL);
+       SoupCacheInputStream *istream = g_object_new (SOUP_TYPE_CACHE_INPUT_STREAM,
+                                             "base-stream", base_stream,
+                                             "close-base-stream", FALSE,
+                                             NULL);
+
+       istream->priv->cancellable = g_cancellable_new ();
+       g_file_replace_async (file, NULL, FALSE,
+                             G_FILE_CREATE_PRIVATE | G_FILE_CREATE_REPLACE_DESTINATION,
+                             G_PRIORITY_DEFAULT, istream->priv->cancellable,
+                             file_replaced_cb, g_object_ref (istream));
+
+       return (GInputStream *) istream;
 }
index c999d10..92b1d7b 100644 (file)
@@ -24,6 +24,9 @@ typedef struct _SoupCacheInputStreamPrivate SoupCacheInputStreamPrivate;
 struct _SoupCacheInputStreamClass
 {
        SoupFilterInputStreamClass parent_class;
+
+       /* signals */
+       void     (*caching_finished) (SoupCacheInputStream *istream, gsize bytes_written, GError *error);
 };
 
 struct _SoupCacheInputStream
@@ -35,17 +38,8 @@ struct _SoupCacheInputStream
 
 GType soup_cache_input_stream_get_type (void) G_GNUC_CONST;
 
-GInputStream *soup_cache_input_stream_new               (GInputStream *base_stream);
-
-void          soup_cache_input_stream_cache             (SoupCacheInputStream  *istream,
-                                                        GFile                 *file,
-                                                        GCancellable          *cancellable,
-                                                        GAsyncReadyCallback    callback,
-                                                        gpointer               user_data);
-
-gsize         soup_cache_input_stream_cache_finish      (SoupCacheInputStream  *istream,
-                                                        GAsyncResult          *result,
-                                                        GError               **error);
+GInputStream *soup_cache_input_stream_new (GInputStream *base_stream,
+                                          GFile        *file);
 
 G_END_DECLS
 
index 839cdc1..db74831 100644 (file)
@@ -749,21 +749,20 @@ typedef struct {
 } StreamHelper;
 
 static void
-istream_cache_cb (GObject *source,
-                 GAsyncResult *res,
-                 gpointer user_data)
+istream_caching_finished (SoupCacheInputStream *istream,
+                         gsize                 bytes_written,
+                         GError               *error,
+                         gpointer              user_data)
 {
-       SoupCacheInputStream *istream = SOUP_CACHE_INPUT_STREAM (source);
        StreamHelper *helper = (StreamHelper *) user_data;
        SoupCache *cache = helper->cache;
        SoupCacheEntry *entry = helper->entry;
-       GError *error = NULL;
 
-       entry->dirty = FALSE;
-       g_clear_object (&entry->cancellable);
        --cache->priv->n_pending;
 
-       entry->length = soup_cache_input_stream_cache_finish (istream, res, &error);
+       entry->dirty = FALSE;
+       entry->length = bytes_written;
+       g_clear_object (&entry->cancellable);
 
        if (error) {
                /* Update cache size */
@@ -791,7 +790,6 @@ istream_cache_cb (GObject *source,
        g_slice_free (StreamHelper, helper);
 }
 
-
 static GInputStream*
 soup_cache_content_processor_wrap_input (SoupContentProcessor *processor,
                                         GInputStream *base_stream,
@@ -853,21 +851,19 @@ soup_cache_content_processor_wrap_input (SoupContentProcessor *processor,
                return NULL;
        }
 
-       ++cache->priv->n_pending;
-
-       istream = soup_cache_input_stream_new (base_stream);
-
-       file = get_file_from_entry (cache, entry);
        entry->cancellable = g_cancellable_new ();
+       ++cache->priv->n_pending;
 
        helper = g_slice_new (StreamHelper);
        helper->cache = g_object_ref (cache);
        helper->entry = entry;
 
-       soup_cache_input_stream_cache (SOUP_CACHE_INPUT_STREAM (istream), file, entry->cancellable,
-                                      (GAsyncReadyCallback) istream_cache_cb, helper);
+       file = get_file_from_entry (cache, entry);
+       istream = soup_cache_input_stream_new (base_stream, file);
        g_object_unref (file);
 
+       g_signal_connect (istream, "caching-finished", G_CALLBACK (istream_caching_finished), helper);
+
        return istream;
 }