HTTPServer: start an I/O loop in a new thread and accept connections

diff --git a/src/httpserver.cpp b/src/httpserver.cpp
index 0b846523..c155544d 100644
--- a/src/httpserver.cpp
+++ b/src/httpserver.cpp
@@ -19,6 +19,7 @@
 #include <util/signalinterrupt.h>
 #include <util/sock.h>
 #include <util/strencodings.h>
+#include <util/thread.h>
 #include <util/threadnames.h>
 #include <util/threadpool.h>
 #include <util/translation.h>
@@ -49,6 +50,10 @@
 
 #include <support/events.h>
 
+//! The set of sockets cannot be modified while waiting, so
+//! the sleep time needs to be small to avoid new sockets stalling.
+static constexpr auto SELECT_TIMEOUT{50ms};
+
 //! Explicit alias for setting socket option methods.
 static constexpr int SOCKET_OPTION_TRUE{1};
 
@@ -977,6 +982,32 @@ void HTTPServer::StopListening()
     m_listen.clear();
 }
 
+void HTTPServer::StartSocketsThreads()
+{
+    m_thread_socket_handler = std::thread(&util::TraceThread,
+                                          "http",
+                                          [this] { ThreadSocketHandler(); });
+}
+
+void HTTPServer::JoinSocketsThreads()
+{
+    if (m_thread_socket_handler.joinable()) {
+        m_thread_socket_handler.join();
+    }
+}
+
+bool HTTPServer::CloseConnection(const HTTPRemoteClient& http_client)
+{
+    size_t erased = std::erase_if(m_connected,
+                                  [&](auto& client){ return client->m_id == http_client.m_id; });
+    if (erased > 0) {
+         // Report back to the main thread
+         m_connected_size.fetch_sub(erased, std::memory_order_relaxed);
+         return true;
+    }
+    return false;
+}
+
 std::unique_ptr<Sock> HTTPServer::AcceptConnection(const Sock& listen_sock, CService& addr)
 {
     // Make sure we only operate on our own listening sockets
@@ -1013,4 +1044,95 @@ HTTPServer::Id HTTPServer::GetNewId()
 {
     return m_next_id.fetch_add(1, std::memory_order_relaxed);
 }
+
+void HTTPServer::NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& addr)
+{
+    if (!sock->IsSelectable()) {
+        LogDebug(BCLog::HTTP,
+                 "connection from %s dropped: non-selectable socket",
+                 addr.ToStringAddrPort());
+        return;
+    }
+
+    // According to the internet TCP_NODELAY is not carried into accepted sockets
+    // on all platforms.  Set it again here just to be sure.
+    if (sock->SetSockOpt(IPPROTO_TCP, TCP_NODELAY, &SOCKET_OPTION_TRUE, sizeof(SOCKET_OPTION_TRUE)) == SOCKET_ERROR) {
+        LogDebug(BCLog::HTTP, "connection from %s: unable to set TCP_NODELAY, continuing anyway",
+                 addr.ToStringAddrPort());
+    }
+
+    const Id id{GetNewId()};
+
+    m_connected.push_back(std::make_shared<HTTPRemoteClient>(id, addr, std::move(sock)));
+    // Report back to the main thread
+    m_connected_size.fetch_add(1, std::memory_order_relaxed);
+
+    LogDebug(BCLog::HTTP,
+             "HTTP Connection accepted from %s (id=%llu)",
+             addr.ToStringAddrPort(), id);
+}
+
+void HTTPServer::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock)
+{
+    for (const auto& sock : m_listen) {
+        if (m_interrupt_net) {
+            return;
+        }
+        const auto it = events_per_sock.find(sock);
+        if (it != events_per_sock.end() && it->second.occurred & Sock::RECV) {
+            CService addr_accepted;
+
+            auto sock_accepted{AcceptConnection(*sock, addr_accepted)};
+
+            if (sock_accepted) {
+                NewSockAccepted(std::move(sock_accepted), addr_accepted);
+            }
+        }
+    }
+}
+
+HTTPServer::IOReadiness HTTPServer::GenerateWaitSockets() const
+{
+    IOReadiness io_readiness;
+
+    for (const auto& sock : m_listen) {
+        io_readiness.events_per_sock.emplace(sock, Sock::Events{Sock::RECV});
+    }
+
+    for (const auto& http_client : m_connected) {
+        // TODO: Implement HTTPRemoteClient methods for send/receive readiness
+        const bool select_recv{true};
+        const bool select_send{true};
+        if (!select_recv && !select_send) continue;
+
+        // Safely copy the shared pointer to the socket
+        std::shared_ptr<Sock> sock{WITH_LOCK(http_client->m_sock_mutex, return http_client->m_sock;)};
+
+        Sock::Event event = (select_send ? Sock::SEND : 0) | (select_recv ? Sock::RECV : 0);
+        io_readiness.events_per_sock.emplace(sock, Sock::Events{event});
+        io_readiness.httpclients_per_sock.emplace(sock, http_client);
+    }
+
+    return io_readiness;
+}
+
+void HTTPServer::ThreadSocketHandler()
+{
+    while (!m_interrupt_net) {
+        // Check for the readiness of the already connected sockets and the
+        // listening sockets in one call ("readiness" as in poll(2) or
+        // select(2)). If none are ready, wait for a short while and return
+        // empty sets.
+        auto io_readiness{GenerateWaitSockets()};
+        if (io_readiness.events_per_sock.empty() ||
+            // WaitMany() may as well be a static method, the context of the first Sock in the vector is not relevant.
+            !io_readiness.events_per_sock.begin()->first->WaitMany(SELECT_TIMEOUT,
+                                                                   io_readiness.events_per_sock)) {
+            m_interrupt_net.sleep_for(SELECT_TIMEOUT);
+        }
+
+        // Accept new connections from listening sockets.
+        SocketHandlerListening(io_readiness.events_per_sock);
+    }
+}
 } // namespace http_bitcoin
diff --git a/src/httpserver.h b/src/httpserver.h
index 5b7922df..73a06441 100644
--- a/src/httpserver.h
+++ b/src/httpserver.h
@@ -21,6 +21,7 @@
 #include <util/sock.h>
 #include <util/strencodings.h>
 #include <util/string.h>
+#include <util/threadinterrupt.h>
 
 namespace util {
 class SignalInterrupt;
@@ -304,6 +305,8 @@ public:
     /// @}
 };
 
+class HTTPRemoteClient;
+
 class HTTPServer
 {
 public:
@@ -312,6 +315,13 @@ public:
      */
     using Id = uint64_t;
 
+    virtual ~HTTPServer()
+    {
+        Assume(!m_thread_socket_handler.joinable()); // Missing call to JoinSocketsThreads()
+        Assume(m_connected.empty()); // Missing call to CloseConnection()
+        Assume(m_listen.empty()); // Missing call to StopListening()
+    }
+
     /**
      * Bind to a new address:port, start listening and add the listen socket to `m_listen`.
      * @param[in] to Where to bind.
@@ -330,14 +340,40 @@ public:
     size_t GetListeningSocketCount() const { return m_listen.size(); }
 
     /**
-     * This is a temporary method used to accept connections from a listening
-     * socket in the unit tests before the I/O loop is implemented.
+     * Get the number of HTTPRemoteClients we are connected to
+     */
+    size_t GetConnectionsCount() const { return m_connected_size.load(std::memory_order_acquire); }
+
+    /**
+     * This is a temporary method used to get a pointer to an HTTPRemoteClient
+     * so its connection can be inspected in a unit test.
      * It will be removed in a future commit.
      */
-    std::unique_ptr<Sock> AcceptConnectionFromListeningSocket(CService& addr)
-    {
-        return AcceptConnection(*m_listen.front(), addr);
-    }
+    const HTTPRemoteClient& GetFirstConnection() { return *m_connected.front(); }
+
+    /**
+     * Start the necessary threads for sockets IO.
+     */
+    void StartSocketsThreads();
+
+    /**
+     * Join (wait for) the threads started by `StartSocketsThreads()` to exit.
+     */
+    void JoinSocketsThreads();
+
+    /**
+     * Stop network activity
+     */
+    void InterruptNet() { m_interrupt_net(); }
+
+    /**
+     * Destroy a given connection by closing its socket and release resources occupied by it.
+     * This is a temporary method that will be removed in a future commit when it is
+     * replaced by DisconnectClients() and run automatically in the I/O thread.
+     * @param[in] http_client Connection to destroy.
+     * @return Whether the connection existed and its socket was closed by this call.
+     */
+    bool CloseConnection(const HTTPRemoteClient& http_client);
 
 private:
     /**
@@ -350,6 +386,56 @@ private:
      */
     std::atomic<Id> m_next_id{0};
 
+    /**
+     * List of HTTPRemoteClients with connected sockets.
+     * Connections will only be added and removed in the I/O thread, but
+     * shared pointers may be passed to worker threads to handle requests
+     * and send replies.
+     */
+    std::vector<std::shared_ptr<HTTPRemoteClient>> m_connected;
+
+    /**
+     * The number of connected sockets.
+     * Updated from the I/O thread but safely readable from
+     * the main thread without locks.
+     */
+    std::atomic<size_t> m_connected_size{0};
+
+    /**
+     * Info about which socket has which event ready and a reverse map
+     * back to the HTTPRemoteClient that owns the socket.
+     */
+    struct IOReadiness {
+        /**
+         * Map of socket -> socket events. For example:
+         * socket1 -> { requested = SEND|RECV, occurred = RECV }
+         * socket2 -> { requested = SEND, occurred = SEND }
+         */
+        Sock::EventsPerSock events_per_sock;
+
+        /**
+         * Map of socket -> HTTPRemoteClient. For example:
+         * socket1 -> HTTPRemoteClient{ id=23 }
+         * socket2 -> HTTPRemoteClient{ id=56 }
+         */
+        std::unordered_map<Sock::EventsPerSock::key_type,
+                           std::shared_ptr<HTTPRemoteClient>,
+                           Sock::HashSharedPtrSock,
+                           Sock::EqualSharedPtrSock>
+            httpclients_per_sock;
+    };
+
+    /**
+     * This is signaled when network activity should cease.
+     */
+    CThreadInterrupt m_interrupt_net;
+
+    /**
+     * Thread that sends to and receives from sockets and accepts connections.
+     * Executes the I/O loop of the server.
+     */
+    std::thread m_thread_socket_handler;
+
     /**
      * Accept a connection.
      * @param[in] listen_sock Socket on which to accept the connection.
@@ -362,6 +448,33 @@ private:
      * Generate an id for a newly created connection.
      */
     Id GetNewId();
+
+    /**
+     * After a new socket with a client has been created, configure its flags,
+     * make a new HTTPRemoteClient and Id and save its shared pointer.
+     * @param[in] sock The newly created socket.
+     * @param[in] addr Address of the new peer.
+     */
+    void NewSockAccepted(std::unique_ptr<Sock>&& sock, const CService& addr);
+
+    /**
+     * Accept incoming connections, one from each read-ready listening socket.
+     * @param[in] events_per_sock Sockets that are ready for IO.
+     */
+    void SocketHandlerListening(const Sock::EventsPerSock& events_per_sock);
+
+    /**
+     * Generate a collection of sockets to check for IO readiness.
+     * @return Sockets to check for readiness plus an aux map to find the
+     * corresponding HTTPRemoteClient given a socket.
+     */
+    IOReadiness GenerateWaitSockets() const;
+
+    /**
+     * Check connected and listening sockets for IO readiness and process them accordingly.
+     * This is the main I/O loop of the server.
+     */
+    void ThreadSocketHandler();
 };
 
 class HTTPRemoteClient
@@ -376,8 +489,24 @@ public:
     //! IP:port of connected client, cached for logging purposes
     const std::string m_origin;
 
-    explicit HTTPRemoteClient(HTTPServer::Id id, const CService& addr)
-        : m_id(id), m_addr(addr), m_origin(addr.ToStringAddrPort()) {};
+    /**
+     * Mutex that serializes the Send() and Recv() calls on `m_sock`. Reading
+     * from the client occurs in the I/O thread but writing back to a client
+     * may occur in a worker thread.
+     */
+    Mutex m_sock_mutex;
+
+    /**
+     * Underlying socket.
+     * `shared_ptr` (instead of `unique_ptr`) is used to avoid premature close of the
+     * underlying file descriptor by one thread while another thread is poll(2)-ing
+     * it for activity.
+     * @see https://github.com/bitcoin/bitcoin/issues/21744 for details.
+     */
+    std::shared_ptr<Sock> m_sock GUARDED_BY(m_sock_mutex);
+
+    explicit HTTPRemoteClient(HTTPServer::Id id, const CService& addr, std::unique_ptr<Sock> socket)
+        : m_id(id), m_addr(addr), m_origin(addr.ToStringAddrPort()), m_sock{std::move(socket)} {};
 
     // Disable copies (should only be used as shared pointers)
     HTTPRemoteClient(const HTTPRemoteClient&) = delete;
diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp
index 22af3981..b30a76f0 100644
--- a/src/test/httpserver_tests.cpp
+++ b/src/test/httpserver_tests.cpp
@@ -11,6 +11,7 @@
 #include <boost/test/unit_test.hpp>
 
 using http_bitcoin::HTTPHeaders;
+using http_bitcoin::HTTPRemoteClient;
 using http_bitcoin::HTTPRequest;
 using http_bitcoin::HTTPResponse;
 using http_bitcoin::HTTPServer;
@@ -396,14 +397,35 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
     // We are bound and listening
     BOOST_REQUIRE_EQUAL(server.GetListeningSocketCount(), 1);
 
-    // Pick up the phone, there's no one there
-    CService addr_connection;
-    BOOST_REQUIRE(!server.AcceptConnectionFromListeningSocket(addr_connection));
+    // Start the I/O loop
+    server.StartSocketsThreads();
+
+    // No connections yet
+    BOOST_CHECK_EQUAL(server.GetConnectionsCount(), 0);
 
     // Create a mock client and add it to the local CreateSock queue
     ConnectClient();
-    // Accept the connection
-    BOOST_REQUIRE(server.AcceptConnectionFromListeningSocket(addr_connection));
-    BOOST_CHECK_EQUAL(addr_connection.ToStringAddrPort(), "5.5.5.5:6789");
+
+    // Wait up to a minute to find and connect the client in the I/O loop
+    int attempts{6000};
+    while (server.GetConnectionsCount() < 1) {
+        std::this_thread::sleep_for(10ms);
+        BOOST_REQUIRE(--attempts > 0);
+    }
+
+    // Inspect the connection
+    const HTTPRemoteClient& client = server.GetFirstConnection();
+    BOOST_CHECK_EQUAL(client.m_origin, "5.5.5.5:6789");
+
+    // Stop the I/O thread before modifying m_connected. CloseConnection() is a
+    // temporary test-only API that is not thread-safe against GenerateWaitSockets(),
+    // which iterates m_connected on the I/O thread.
+    server.InterruptNet();
+    server.JoinSocketsThreads();
+    // Close connection
+    BOOST_REQUIRE(server.CloseConnection(client));
+    // Close all listening sockets
+    server.StopListening();
 }
+
 BOOST_AUTO_TEST_SUITE_END()