diff --git a/src/httpserver.cpp b/src/httpserver.cpp index 0b846523711..c155544dc9a 100644 --- a/src/httpserver.cpp +++ b/src/httpserver.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -49,6 +50,10 @@ #include +//! 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 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, 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(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{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 5b7922dfa14..73a06441a80 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -21,6 +21,7 @@ #include #include #include +#include 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 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 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> m_connected; + + /** + * The number of connected sockets. + * Updated from the I/O thread but safely readable from + * the main thread without locks. + */ + std::atomic 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::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, 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 m_sock GUARDED_BY(m_sock_mutex); + + explicit HTTPRemoteClient(HTTPServer::Id id, const CService& addr, std::unique_ptr 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 22af3981f2d..b30a76f0458 100644 --- a/src/test/httpserver_tests.cpp +++ b/src/test/httpserver_tests.cpp @@ -11,6 +11,7 @@ #include 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()