[ostree] fetcher: Add a queue to limit requests sent to soup

[Sjoerd Simons <sjoerds src gnome org>]

commit 5f310868f778d3691fcd269718258a8680394a01
Author: Sjoerd Simons <sjoerd luon net>
Date:   Sun Sep 15 21:25:02 2013 +0200

    fetcher: Add a queue to limit requests sent to soup
    
    On a large ostree repository pulling over http slows to a crawl. Pulling
    from localhost results in:
     5944 metadata, 63734 content objects fetched; 850509 KiB transferred in
     1106 seconds
    In other words about 800KiB/s. Some profiling shows that essentially
    all of the CPU goes into libsoup doing its request bookkeeping instead
    of into the actual downloading.
    
    Adding a simple queue to limit to number of active request sent into
    libsoup makes for a dramatic improvement:
     5944 metadata, 63734 content objects fetched; 850509 KiB transferred
     in 89 seconds
    So around 9450 KiB/s.
    
    https://bugzilla.gnome.org/show_bug.cgi?id=708126

 src/libostree/ostree-fetcher.c |   64 ++++++++++++++++++++++++++++++++++++++--
 1 files changed, 61 insertions(+), 3 deletions(-)
---
diff --git a/src/libostree/ostree-fetcher.c b/src/libostree/ostree-fetcher.c
index 519345f..3316dd9 100644
--- a/src/libostree/ostree-fetcher.c
+++ b/src/libostree/ostree-fetcher.c
@@ -53,6 +53,9 @@ typedef struct {
   GSimpleAsyncResult *result;
 } OstreeFetcherPendingURI;
 
+static void ostree_fetcher_pending_uri_done (OstreeFetcher *self,
+                                             OstreeFetcherPendingURI *pending);
+
 static void
 pending_uri_free (OstreeFetcherPendingURI *pending)
 {
@@ -61,6 +64,9 @@ pending_uri_free (OstreeFetcherPendingURI *pending)
   if (pending->refcount > 0)
     return;
 
+  if (!pending->is_stream)
+    ostree_fetcher_pending_uri_done (pending->self, pending);
+
   soup_uri_free (pending->uri);
   g_clear_object (&pending->self);
   g_clear_object (&pending->out_tmpfile);
@@ -86,6 +92,11 @@ struct OstreeFetcher
   
   guint64 total_downloaded;
   guint total_requests;
+
+  /* Queue for libsoup, see bgo#708591 */
+  gint outstanding;
+  GQueue pending_queue;
+  gint max_outstanding;
 };
 
 G_DEFINE_TYPE (OstreeFetcher, ostree_fetcher, G_TYPE_OBJECT)
@@ -103,6 +114,8 @@ ostree_fetcher_finalize (GObject *object)
   g_hash_table_destroy (self->sending_messages);
   g_hash_table_destroy (self->message_to_request);
 
+  g_queue_clear (&self->pending_queue);
+
   G_OBJECT_CLASS (ostree_fetcher_parent_class)->finalize (object);
 }
 
@@ -137,12 +150,17 @@ on_request_unqueued (SoupSession  *session,
 static void
 ostree_fetcher_init (OstreeFetcher *self)
 {
+  gint max_conns;
+
+  g_queue_init (&self->pending_queue);
   self->session = soup_session_async_new_with_options (SOUP_SESSION_USER_AGENT, "ostree ",
                                                        SOUP_SESSION_SSL_USE_SYSTEM_CA_FILE, TRUE,
                                                        SOUP_SESSION_USE_THREAD_CONTEXT, TRUE,
                                                        SOUP_SESSION_ADD_FEATURE_BY_TYPE, SOUP_TYPE_REQUESTER,
                                                        NULL);
   self->requester = (SoupRequester *)soup_session_get_feature (self->session, SOUP_TYPE_REQUESTER);
+  g_object_get (self->session, "max-conns-per-host", &max_conns, NULL);
+  self->max_outstanding = 3 * max_conns;
 
   g_signal_connect (self->session, "request-started",
                     G_CALLBACK (on_request_started), self);
@@ -169,6 +187,47 @@ ostree_fetcher_new (GFile                    *tmpdir,
 }
 
 static void
+on_request_sent (GObject        *object, GAsyncResult   *result, gpointer        user_data);
+
+static void
+ostree_fetcher_pending_uri_done (OstreeFetcher *self,
+                                 OstreeFetcherPendingURI *pending)
+{
+  OstreeFetcherPendingURI *p;
+
+  g_assert (!pending->is_stream);
+
+  self->outstanding--;
+  p = g_queue_pop_head (&self->pending_queue);
+  if (p != NULL)
+    {
+      self->outstanding++;
+      soup_request_send_async (p->request, p->cancellable,
+                           on_request_sent, p);
+    }
+}
+
+static void
+ostree_fetcher_queue_pending_uri (OstreeFetcher *self,
+                                  OstreeFetcherPendingURI *pending)
+{
+  g_assert (!pending->is_stream);
+
+  if (self->outstanding >= self->max_outstanding)
+    {
+      g_queue_push_tail (&self->pending_queue, pending);
+    }
+  else
+    {
+      self->outstanding++;
+      soup_request_send_async (pending->request, pending->cancellable,
+                           on_request_sent, pending);
+    }
+}
+
+
+
+static void
 on_splice_complete (GObject        *object,
                     GAsyncResult   *result,
                     gpointer        user_data) 
@@ -351,9 +410,8 @@ ostree_fetcher_request_uri_with_partial_async (OstreeFetcher         *self,
                            soup_request_http_get_message ((SoupRequestHTTP*)pending->request),
                            pending);
     }
-  
-  soup_request_send_async (pending->request, cancellable,
-                           on_request_sent, pending);
+
+  ostree_fetcher_queue_pending_uri (self, pending);
 
  out:
   if (local_error != NULL)