RequestServer+LibHTTP+LibGemini: Cache connections to the same host

This makes connections (particularly TLS-based ones) do the handshaking
stuff only once.
Currently the cache is configured to keep at most two connections evenly
balanced in queue size, and with a grace period of 10s after the last
queued job has finished (after which the connection will be dropped).

Userland/Services/RequestServer/CMakeLists.txt

@@ -8,6 +8,7 @@ compile_ipc(RequestClient.ipc RequestClientEndpoint.h)
 
 set(SOURCES
     ClientConnection.cpp
+    ConnectionCache.cpp
     Request.cpp
     RequestClientEndpoint.h
     RequestServerEndpoint.h

Userland/Services/RequestServer/ConnectionCache.cpp

@@ -0,0 +1,73 @@
+/*
+ * Copyright (c) 2021, Ali Mohammad Pur <mpfard@serenityos.org>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#include "ConnectionCache.h"
+#include <LibCore/EventLoop.h>
+
+namespace RequestServer::ConnectionCache {
+
+HashMap<ConnectionKey, Vector<Connection<Core::TCPSocket>>> g_tcp_connection_cache {};
+HashMap<ConnectionKey, Vector<Connection<TLS::TLSv12>>> g_tls_connection_cache {};
+
+void request_did_finish(URL const& url, Core::Socket const* socket)
+{
+    if (!socket) {
+        dbgln("Request with a null socket finished for URL {}", url);
+        return;
+    }
+
+    dbgln("Request for {} finished", url);
+
+    ConnectionKey key { url.host(), url.port_or_default() };
+    auto fire_off_next_job = [&](auto& cache) {
+        auto it = cache.find(key);
+        if (it == cache.end()) {
+            dbgln("Request for URL {} finished, but we don't own that!", url);
+            return;
+        }
+        auto connection_it = it->value.find_if([&](auto& connection) { return connection.socket == socket; });
+        if (connection_it.is_end()) {
+            dbgln("Request for URL {} finished, but we don't have a socket for that!", url);
+            return;
+        }
+
+        auto& connection = *connection_it;
+        if (connection.request_queue.is_empty()) {
+            connection.has_started = false;
+            connection.removal_timer->on_timeout = [&connection, &cache_entry = it->value] {
+                Core::deferred_invoke([&] {
+                    dbgln("Removing no-longer-used connection {}", &connection);
+                    cache_entry.remove_first_matching([&](auto& entry) { return &entry == &connection; });
+                });
+            };
+            connection.removal_timer->start();
+        } else {
+            using SocketType = RemoveCVReference<decltype(*connection.socket)>;
+            bool is_connected;
+            if constexpr (IsSame<SocketType, TLS::TLSv12>)
+                is_connected = connection.socket->is_established();
+            else
+                is_connected = connection.socket->is_connected();
+            if (!is_connected) {
+                // Create another socket for the connection.
+                dbgln("Creating a new socket for {}", url);
+                connection.socket = SocketType::construct(nullptr);
+            }
+            dbgln("Running next job in queue for connection {}", &connection);
+            auto request = connection.request_queue.take_first();
+            request(connection.socket);
+        }
+    };
+
+    if (is<TLS::TLSv12>(socket))
+        fire_off_next_job(g_tls_connection_cache);
+    else if (is<Core::TCPSocket>(socket))
+        fire_off_next_job(g_tcp_connection_cache);
+    else
+        dbgln("Unknown socket {} finished for URL {}", *socket, url);
+}
+
+}

Userland/Services/RequestServer/ConnectionCache.h

@@ -0,0 +1,103 @@
+/*
+ * Copyright (c) 2021, Ali Mohammad Pur <mpfard@serenityos.org>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <AK/HashMap.h>
+#include <AK/URL.h>
+#include <AK/Vector.h>
+#include <LibCore/TCPSocket.h>
+#include <LibCore/Timer.h>
+#include <LibTLS/TLSv12.h>
+
+namespace RequestServer::ConnectionCache {
+
+template<typename Socket>
+struct Connection {
+    using QueueType = Vector<Function<void(Core::Socket&)>>;
+    using SocketType = Socket;
+
+    NonnullRefPtr<Socket> socket;
+    QueueType request_queue;
+    NonnullRefPtr<Core::Timer> removal_timer;
+    bool has_started { false };
+};
+
+struct ConnectionKey {
+    String hostname;
+    u16 port { 0 };
+
+    bool operator==(ConnectionKey const&) const = default;
+};
+
+};
+
+template<>
+struct AK::Traits<RequestServer::ConnectionCache::ConnectionKey> : public AK::GenericTraits<RequestServer::ConnectionCache::ConnectionKey> {
+    static u32 hash(RequestServer::ConnectionCache::ConnectionKey const& key)
+    {
+        return pair_int_hash(key.hostname.hash(), key.port);
+    }
+};
+
+namespace RequestServer::ConnectionCache {
+
+extern HashMap<ConnectionKey, Vector<Connection<Core::TCPSocket>>> g_tcp_connection_cache;
+extern HashMap<ConnectionKey, Vector<Connection<TLS::TLSv12>>> g_tls_connection_cache;
+
+void request_did_finish(URL const&, Core::Socket const*);
+
+constexpr static inline size_t MaxConcurrentConnectionsPerURL = 2;
+constexpr static inline size_t ConnectionKeepAliveTimeMilliseconds = 10'000;
+
+decltype(auto) get_or_create_connection(auto& cache, URL const& url, auto& job)
+{
+    auto start_job = [&job](auto& socket) {
+        job.start(socket);
+    };
+    auto& sockets_for_url = cache.ensure({ url.host(), url.port_or_default() });
+    auto it = sockets_for_url.find_if([](auto& connection) { return connection.request_queue.is_empty(); });
+    auto did_add_new_connection = false;
+    if (it.is_end() && sockets_for_url.size() < ConnectionCache::MaxConcurrentConnectionsPerURL) {
+        sockets_for_url.append({
+            RemoveCVReference<decltype(cache.begin()->value.at(0))>::SocketType::construct(nullptr),
+            {},
+            Core::Timer::create_single_shot(ConnectionKeepAliveTimeMilliseconds, nullptr),
+        });
+        did_add_new_connection = true;
+    }
+    size_t index;
+    if (it.is_end()) {
+        if (did_add_new_connection) {
+            index = sockets_for_url.size() - 1;
+        } else {
+            // Find the least backed-up connection (based on how many entries are in their request queue.
+            index = 0;
+            auto min_queue_size = (size_t)-1;
+            for (auto it = sockets_for_url.begin(); it != sockets_for_url.end(); ++it) {
+                if (auto queue_size = it->request_queue.size(); min_queue_size > queue_size) {
+                    index = it.index();
+                    min_queue_size = queue_size;
+                }
+            }
+        }
+    } else {
+        index = it.index();
+    }
+    auto& connection = sockets_for_url[index];
+    if (!connection.has_started) {
+        dbgln("Immediately start request for url {} in {}", url, &connection);
+        connection.has_started = true;
+        connection.removal_timer->stop();
+        start_job(*connection.socket);
+    } else {
+        dbgln("Enqueue request for URL {} in {}", url, &connection);
+        connection.request_queue.append(move(start_job));
+    }
+    return connection;
+}
+
+}

Userland/Services/RequestServer/GeminiProtocol.cpp

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
+#include "ConnectionCache.h"
 #include <LibGemini/GeminiJob.h>
 #include <LibGemini/GeminiRequest.h>
 #include <RequestServer/GeminiProtocol.h>
@@ -34,7 +35,9 @@ OwnPtr<Request> GeminiProtocol::start_request(ClientConnection& client, const St
     auto job = Gemini::GeminiJob::construct(request, *output_stream);
     auto protocol_request = GeminiRequest::create_with_job({}, client, (Gemini::GeminiJob&)*job, move(output_stream));
     protocol_request->set_request_fd(pipe_result.value().read_fd);
-    job->start();
+
+    ConnectionCache::get_or_create_connection(ConnectionCache::g_tls_connection_cache, url, *job);
+
     return protocol_request;
 }
 

Userland/Services/RequestServer/GeminiRequest.cpp

@@ -4,6 +4,8 @@
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
+#include "ConnectionCache.h"
+#include <LibCore/EventLoop.h>
 #include <LibGemini/GeminiJob.h>
 #include <LibGemini/GeminiResponse.h>
 #include <RequestServer/GeminiRequest.h>
@@ -15,6 +17,9 @@ GeminiRequest::GeminiRequest(ClientConnection& client, NonnullRefPtr<Gemini::Gem
     , m_job(job)
 {
     m_job->on_finish = [this](bool success) {
+        Core::deferred_invoke([url = m_job->url(), socket = m_job->socket()] {
+            ConnectionCache::request_did_finish(url, socket);
+        });
         if (auto* response = m_job->response()) {
             set_downloaded_size(this->output_stream().size());
             if (!response->meta().is_empty()) {

Userland/Services/RequestServer/GeminiRequest.h

@@ -18,6 +18,8 @@ public:
     virtual ~GeminiRequest() override;
     static NonnullOwnPtr<GeminiRequest> create_with_job(Badge<GeminiProtocol>, ClientConnection&, NonnullRefPtr<Gemini::GeminiJob>, NonnullOwnPtr<OutputFileStream>&&);
 
+    Gemini::GeminiJob const& job() const { return *m_job; }
+
 private:
     explicit GeminiRequest(ClientConnection&, NonnullRefPtr<Gemini::GeminiJob>, NonnullOwnPtr<OutputFileStream>&&);
 

Userland/Services/RequestServer/HttpCommon.h

@@ -15,6 +15,7 @@
 #include <AK/Types.h>
 #include <LibHTTP/HttpRequest.h>
 #include <RequestServer/ClientConnection.h>
+#include <RequestServer/ConnectionCache.h>
 #include <RequestServer/Request.h>
 
 namespace RequestServer::Detail {
@@ -29,6 +30,9 @@ void init(TSelf* self, TJob job)
     };
 
     job->on_finish = [self](bool success) {
+        Core::deferred_invoke([url = self->job().url(), socket = self->job().socket()] {
+            ConnectionCache::request_did_finish(url, socket);
+        });
         if (auto* response = self->job().response()) {
             self->set_status_code(response->code());
             self->set_response_headers(response->headers());
@@ -80,7 +84,12 @@ OwnPtr<Request> start_request(TBadgedProtocol&& protocol, ClientConnection& clie
     auto job = TJob::construct(request, *output_stream);
     auto protocol_request = TRequest::create_with_job(forward<TBadgedProtocol>(protocol), client, (TJob&)*job, move(output_stream));
     protocol_request->set_request_fd(pipe_result.value().read_fd);
-    job->start();
+
+    if constexpr (IsSame<typename TBadgedProtocol::Type, HttpsProtocol>)
+        ConnectionCache::get_or_create_connection(ConnectionCache::g_tls_connection_cache, url, *job);
+    else
+        ConnectionCache::get_or_create_connection(ConnectionCache::g_tcp_connection_cache, url, *job);
+
     return protocol_request;
 }
 

Userland/Services/RequestServer/HttpRequest.h

@@ -20,6 +20,7 @@ public:
     static NonnullOwnPtr<HttpRequest> create_with_job(Badge<HttpProtocol>&&, ClientConnection&, NonnullRefPtr<HTTP::HttpJob>, NonnullOwnPtr<OutputFileStream>&&);
 
     HTTP::HttpJob& job() { return m_job; }
+    HTTP::HttpJob const& job() const { return m_job; }
 
 private:
     explicit HttpRequest(ClientConnection&, NonnullRefPtr<HTTP::HttpJob>, NonnullOwnPtr<OutputFileStream>&&);

Userland/Services/RequestServer/HttpsRequest.h

@@ -19,6 +19,7 @@ public:
     static NonnullOwnPtr<HttpsRequest> create_with_job(Badge<HttpsProtocol>&&, ClientConnection&, NonnullRefPtr<HTTP::HttpsJob>, NonnullOwnPtr<OutputFileStream>&&);
 
     HTTP::HttpsJob& job() { return m_job; }
+    HTTP::HttpsJob const& job() const { return m_job; }
 
 private:
     explicit HttpsRequest(ClientConnection&, NonnullRefPtr<HTTP::HttpsJob>, NonnullOwnPtr<OutputFileStream>&&);