From e5f242eef3a1eb86760b694f83921f460046e004 Mon Sep 17 00:00:00 2001 From: Matthew Zipkin Date: Wed, 15 Jan 2025 15:17:36 -0500 Subject: [PATCH] HTTPServer: implement control methods to match legacy API --- src/httpserver.cpp | 116 +++++++++++++++++++++++++++++++++++++++++++++ src/httpserver.h | 68 +++++++++++++++++++++++--- 2 files changed, 178 insertions(+), 6 deletions(-) diff --git a/src/httpserver.cpp b/src/httpserver.cpp index e912e854227..09dec7ae9a7 100644 --- a/src/httpserver.cpp +++ b/src/httpserver.cpp @@ -82,6 +82,7 @@ struct HTTPPathHandler static struct event_base* eventBase = nullptr; //! HTTP server static struct evhttp* eventHTTP = nullptr; +static std::unique_ptr g_http_server{nullptr}; //! List of subnets to allow RPC connections from static std::vector rpc_allow_subnets; //! Handlers for (sub)paths @@ -274,6 +275,12 @@ static void MaybeDispatchRequestToWorker(std::shared_ptr hreq) } } +static void RejectRequest(std::unique_ptr hreq) +{ + LogDebug(BCLog::HTTP, "Rejecting request while shutting down"); + hreq->WriteReply(HTTP_SERVICE_UNAVAILABLE); +} + /** HTTP request callback */ static void http_request_cb(struct evhttp_request* req, void* arg) { @@ -1373,6 +1380,7 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const void HTTPServer::SocketHandlerListening(const Sock::EventsPerSock& events_per_sock) { + if (m_stop_accepting) return; for (const auto& sock : m_listen) { if (m_interrupt_net) { return; @@ -1492,6 +1500,7 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptrm_req_queue.empty()) { + LOCK(m_request_dispatcher_mutex); client->m_req_busy = true; m_request_dispatcher(std::move(client->m_req_queue.front())); client->m_req_queue.pop_front(); @@ -1538,6 +1547,15 @@ void HTTPServer::DisconnectClients() } } +void HTTPServer::ClearConnectedClients() +{ + Assume(!m_thread_socket_handler.joinable()); // must be called after JoinSocketsThreads() + if (m_connected.empty()) return; + LogWarning("Force-disconnecting %d HTTP client(s) that did not disconnect gracefully", m_connected.size()); + m_connected_size.fetch_sub(m_connected.size(), std::memory_order_relaxed); + m_connected.clear(); +} + bool HTTPRemoteClient::ReadRequest(HTTPRequest& req) { LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE); @@ -1638,4 +1656,102 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer() return true; } + +bool InitHTTPServer() +{ + if (!InitHTTPAllowList()) { + return false; + } + + // Create HTTPServer, using a dummy request handler just for this commit + g_http_server = std::make_unique([&](std::unique_ptr req){}); + + // Bind HTTP server to specified addresses + std::vector> endpoints{GetBindAddresses()}; + bool bind_success{false}; + for (const auto& [address_string, port] : endpoints) { + LogInfo("Binding RPC on address %s port %i", address_string, port); + const std::optional addr{Lookup(address_string, port, false)}; + if (addr) { + if (addr->IsBindAny()) { + LogWarning("The RPC server is not safe to expose to untrusted networks such as the public internet"); + } + auto result{g_http_server->BindAndStartListening(addr.value())}; + if (!result) { + LogWarning("Binding RPC on address %s failed: %s", addr->ToStringAddrPort(), result.error()); + } else { + bind_success = true; + } + } else { + LogWarning("Could not bind RPC on address %s port %i: Address lookup failed.", address_string, port); + } + } + + if (!bind_success) { + LogError("Unable to bind any endpoint for RPC server"); + return false; + } + + LogDebug(BCLog::HTTP, "Initialized HTTP server"); + + g_max_queue_depth = std::max(gArgs.GetArg("-rpcworkqueue", DEFAULT_HTTP_WORKQUEUE), 1); + LogDebug(BCLog::HTTP, "set work queue of depth %d\n", g_max_queue_depth); + + return true; +} + +void StartHTTPServer() +{ + auto rpcThreads{std::max(gArgs.GetArg("-rpcthreads", DEFAULT_HTTP_THREADS), 1)}; + LogInfo("Starting HTTP server with %d worker threads", rpcThreads); + g_threadpool_http.Start(rpcThreads); + g_http_server->StartSocketsThreads(); +} + +void InterruptHTTPServer() +{ + LogDebug(BCLog::HTTP, "Interrupting HTTP server"); + if (g_http_server) { + // Reject all new requests + g_http_server->SetRequestHandler(RejectRequest); + } + + // Interrupt pool after disabling requests + g_threadpool_http.Interrupt(); +} + +void StopHTTPServer() +{ + LogDebug(BCLog::HTTP, "Stopping HTTP server"); + + LogDebug(BCLog::HTTP, "Waiting for HTTP worker threads to exit\n"); + g_threadpool_http.Stop(); + + if (g_http_server) { + // Must precede DisconnectAllClients(): a connection accepted after + // GetConnectionsCount() returns 0 would survive into the destructor. + g_http_server->StopAccepting(); + // Disconnect clients as their remaining responses are flushed + g_http_server->DisconnectAllClients(); + // Wait 30 seconds for all disconnections + LogDebug(BCLog::HTTP, "Waiting for HTTP clients to disconnect gracefully"); + const auto deadline{NodeClock::now() + 30s}; + while (g_http_server->GetConnectionsCount() != 0) { + if (NodeClock::now() > deadline) { + LogWarning("Timeout waiting for HTTP clients to disconnect gracefully, continuing shutdown"); + break; + } + std::this_thread::sleep_for(50ms); + } + // Break HTTPServer I/O loop: stop accepting connections, sending and receiving data + g_http_server->InterruptNet(); + // Wait for HTTPServer I/O thread to exit + g_http_server->JoinSocketsThreads(); + // Force-remove any clients that survived the graceful wait + g_http_server->ClearConnectedClients(); + // Close all listening sockets + g_http_server->StopListening(); + } + LogDebug(BCLog::HTTP, "Stopped HTTP server"); +} } // namespace http_bitcoin diff --git a/src/httpserver.h b/src/httpserver.h index cbd3d215069..a498e400523 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -342,7 +342,8 @@ public: */ using Id = uint64_t; - explicit HTTPServer(std::function&&)> func) : m_request_dispatcher{std::move(func)} {} + explicit HTTPServer(std::function&&)> func) + : m_request_dispatcher{std::move(func)} {} virtual ~HTTPServer() { @@ -393,6 +394,31 @@ public: */ void DisconnectAllClients() { m_disconnect_all_clients = true; } + /** + * Update the request handler method. + * Used for shutdown to reject new requests. + */ + void SetRequestHandler(std::function&&)> func) + EXCLUSIVE_LOCKS_REQUIRED(!m_request_dispatcher_mutex) + { + WITH_LOCK(m_request_dispatcher_mutex, + m_request_dispatcher = std::move(func)); + } + + /** + * Stop accepting new connections in the I/O loop. + * Must be called first in StopHTTPServer() before DisconnectAllClients(). + * A connection accepted after the "wait for 0 connections" loop exits would + * remain in m_connected when the destructor is called. + */ + void StopAccepting() { m_stop_accepting = true; } + + /** + * Force-remove all remaining clients from m_connected without waiting for + * graceful disconnection. Must only be called after JoinSocketsThreads(). + */ + void ClearConnectedClients(); + private: /** * List of listening sockets. @@ -412,6 +438,12 @@ private: */ std::vector> m_connected; + /** + * Flag used during shutdown to stop accepting new connections. + * Set by main thread and read by the I/O thread. + */ + std::atomic_bool m_stop_accepting{false}; + /** * Flag used during shutdown. * Overrides HTTPRemoteClient flags m_keep_alive and m_connection_busy. @@ -462,9 +494,14 @@ private: std::thread m_thread_socket_handler; /* - * What to do with HTTP requests once received, validated and parsed + * What to do with HTTP requests once received, validated and parsed. + * Set in main thread by server start and interrupt but read in + * worker threads. */ - std::function&&)> m_request_dispatcher; + /// @{ + mutable Mutex m_request_dispatcher_mutex; + std::function&&)> m_request_dispatcher GUARDED_BY(m_request_dispatcher_mutex); + /// @} /** * Accept a connection. @@ -491,7 +528,8 @@ private: * Do the read/write for connected sockets that are ready for IO. * @param[in] io_readiness Which sockets are ready and their corresponding HTTPRemoteClients. */ - void SocketHandlerConnected(const IOReadiness& io_readiness) const; + void SocketHandlerConnected(const IOReadiness& io_readiness) const + EXCLUSIVE_LOCKS_REQUIRED(!m_request_dispatcher_mutex); /** * Accept incoming connections, one from each read-ready listening socket. @@ -510,7 +548,7 @@ private: * Check connected and listening sockets for IO readiness and process them accordingly. * This is the main I/O loop of the server. */ - void ThreadSocketHandler(); + void ThreadSocketHandler() EXCLUSIVE_LOCKS_REQUIRED(!m_request_dispatcher_mutex); /** * Try to read HTTPRequests from a client's receive buffer. @@ -519,7 +557,8 @@ private: * will mark this client for disconnection. * @param[in] client The HTTPRemoteClient to read requests from */ - void MaybeDispatchRequestsFromClient(const std::shared_ptr& client) const; + void MaybeDispatchRequestsFromClient(const std::shared_ptr& client) const + EXCLUSIVE_LOCKS_REQUIRED(!m_request_dispatcher_mutex); /** * Close underlying socket connections for flagged clients @@ -633,6 +672,23 @@ public: */ bool MaybeSendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex); }; + +/** Initialize HTTP server. + * Call this before RegisterHTTPHandler or EventBase(). + */ +bool InitHTTPServer(); + +/** Start HTTP server. + * This is separate from InitHTTPServer to give users race-condition-free time + * to register their handlers between InitHTTPServer and StartHTTPServer. + */ +void StartHTTPServer(); + +/** Interrupt HTTP server threads */ +void InterruptHTTPServer(); + +/** Stop HTTP server */ +void StopHTTPServer(); } // namespace http_bitcoin #endif // BITCOIN_HTTPSERVER_H