diff --git a/src/httpserver.cpp b/src/httpserver.cpp index 572fa140258..8467c1f9c52 100644 --- a/src/httpserver.cpp +++ b/src/httpserver.cpp @@ -986,8 +986,7 @@ void HTTPRequest::WriteReply(HTTPStatusCode status, std::span r bool needs_body{status != HTTP_NO_CONTENT && (status < 100 || status >= 200)}; bool needs_content_length{false}; - // Keep track of this, will be used in a future commit to disconnect some clients - [[maybe_unused]] bool keep_alive{false}; + bool keep_alive{false}; // See libevent evhttp_make_header_response() // Expected response headers depend on protocol version @@ -1035,6 +1034,8 @@ void HTTPRequest::WriteReply(HTTPStatusCode status, std::span r keep_alive = false; } + m_client->m_keep_alive = keep_alive; + // Serialize the response headers const std::string headers{res.StringifyHeaders()}; const auto headers_bytes{std::as_bytes(std::span{headers})}; @@ -1158,18 +1159,6 @@ void HTTPServer::JoinSocketsThreads() } } -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 @@ -1276,7 +1265,7 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const client->m_origin, client->m_id, NetworkErrorString(err)); - // TODO: Disconnect + client->m_disconnect = true; } } else if (nrecv == 0) { LogDebug( @@ -1284,8 +1273,11 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const "Received EOF from %s (id=%llu)", client->m_origin, client->m_id); - // TODO: Disconnect + client->m_disconnect = true; } else { + // Prevent disconnect until all requests are completely handled. + client->m_connection_busy = true; + // Copy data from socket buffer to client receive buffer client->m_recv_buffer.insert( client->m_recv_buffer.end(), @@ -1359,6 +1351,9 @@ void HTTPServer::ThreadSocketHandler() // Accept new connections from listening sockets. SocketHandlerListening(io_readiness.events_per_sock); + + // Disconnect any clients that have been flagged. + DisconnectClients(); } } @@ -1371,6 +1366,17 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptrReadRequest(*req)) break; + } catch (const ContentTooLargeError& e) { + LogDebug( + BCLog::HTTP, + "HTTP request body too large from client %s (id=%llu): %s", + client->m_origin, + client->m_id, + e.what()); + + req->WriteReply(HTTP_CONTENT_TOO_LARGE); + client->m_disconnect = true; + return; } catch (const std::runtime_error& e) { LogDebug( BCLog::HTTP, @@ -1380,8 +1386,8 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptrWriteReply(HTTP_BAD_REQUEST); + client->m_disconnect = true; return; } @@ -1399,6 +1405,46 @@ void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptrm_disconnect) { + ; + } else { + // Disconnect this client because the server is shutting + // down and we need to disconnect all clients... + if (m_disconnect_all_clients) { + // ...unless we still have data for this client. + if (client->m_connection_busy) { + // There is still data for this healthy-connected client. + // Continue the I/O loop until all data is sent or an error is encountered. + return false; + } else { + // This is a healthy persistent connection (e.g. keep-alive) + // but it's time to say goodbye. + ; + } + } else { + // No reason to disconnect. + return false; + } + } + // No reason NOT to disconnect, log and remove. + LogDebug(BCLog::HTTP, + "Disconnecting HTTP client %s (id=%llu)", + client->m_origin, + client->m_id); + return true; + }); + if (erased > 0) { + // Report back to the main thread + m_connected_size.fetch_sub(erased, std::memory_order_relaxed); + } +} + bool HTTPRemoteClient::ReadRequest(HTTPRequest& req) { LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE); @@ -1447,6 +1493,7 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer() if (!IOErrorIsPermanent(err)) { // The error can be safely ignored, try the send again on the next I/O loop. m_send_ready = true; + m_connection_busy = true; return true; } else { // Unrecoverable error, log and disconnect client. @@ -1456,7 +1503,9 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer() m_origin, m_id, NetworkErrorString(err)); - // TODO: disconnect + m_send_ready = false; + m_disconnect = true; + // Do not attempt to read from this client. return false; } @@ -1479,9 +1528,14 @@ bool HTTPRemoteClient::MaybeSendBytesFromBuffer() // on an already-empty m_send_buffer because the connection might have just been opened. if (m_send_buffer.empty()) { m_send_ready = false; - } else { - // The send buffer isn't flushed yet, try to push more on the next loop. - m_send_ready = true; + m_connection_busy = false; + + // Our work is done here + if (!m_keep_alive) { + m_disconnect = true; + // Do not attempt to read from this client. + return false; + } } } diff --git a/src/httpserver.h b/src/httpserver.h index 26345c0fdf8..4018837d8ce 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -336,7 +336,7 @@ public: virtual ~HTTPServer() { Assume(!m_thread_socket_handler.joinable()); // Missing call to JoinSocketsThreads() - Assume(m_connected.empty()); // Missing call to CloseConnection() + Assume(m_connected.empty()); // Missing call to DisconnectClients(), or disconnect flags not set Assume(m_listen.empty()); // Missing call to StopListening() } @@ -378,13 +378,9 @@ public: 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. + * Start disconnecting clients when possible in the I/O loop */ - bool CloseConnection(const HTTPRemoteClient& http_client); + void DisconnectAllClients() { m_disconnect_all_clients = true; } private: /** @@ -405,6 +401,13 @@ private: */ std::vector> m_connected; + /** + * Flag used during shutdown. + * Overrides HTTPRemoteClient flags m_keep_alive and m_connection_busy. + * Set by main thread and read by the I/O thread. + */ + std::atomic_bool m_disconnect_all_clients{false}; + /** * The number of connected sockets. * Updated from the I/O thread but safely readable from @@ -506,6 +509,14 @@ private: * @param[in] client The HTTPRemoteClient to read requests from */ void MaybeDispatchRequestsFromClient(const std::shared_ptr& client) const; + + /** + * Close underlying socket connections for flagged clients + * by removing their shared pointer from m_connected. If an HTTPRemoteClient + * is busy in a worker thread, its connection will be closed once that + * job is done and the HTTPRequest is out of scope. + */ + void DisconnectClients(); }; class HTTPRemoteClient @@ -559,6 +570,25 @@ public: */ std::shared_ptr m_sock GUARDED_BY(m_sock_mutex); + //! Initialized to true while server waits for first request from client. + //! Set to false after data is written to m_send_buffer and then that buffer is flushed to client. + //! Reset to true when we receive new request data from client. + //! Checked during DisconnectClients(). All of these operations take place in the HTTPServer I/O loop. + //! `m_connection_busy=true` can be overridden by `m_disconnect=true` (we disconnect). + bool m_connection_busy{true}; + + //! Client has requested to keep the connection open after all requests have been responded to. + //! Set by (potentially multiple) worker threads and checked in the HTTPServer I/O loop. + //! `m_keep_alive=true` can be overridden `by HTTPServer.m_disconnect_all_clients` (we disconnect). + std::atomic_bool m_keep_alive{false}; + + //! Flag this client for disconnection on next loop. + //! Either we have encountered a permanent error, or both sides of the socket are done + //! with the connection, e.g. our reply to a "Connection: close" request has been sent. + //! Might be set in a worker thread or in the I/O thread. When set to `true` we disconnect, + //! possibly overriding all other disconnect flags. + std::atomic_bool m_disconnect{false}; + 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)} {}; diff --git a/src/rpc/protocol.h b/src/rpc/protocol.h index 0ba17ae0218..99e8f05f4c4 100644 --- a/src/rpc/protocol.h +++ b/src/rpc/protocol.h @@ -16,6 +16,7 @@ enum HTTPStatusCode : int HTTP_FORBIDDEN = 403, HTTP_NOT_FOUND = 404, HTTP_BAD_METHOD = 405, + HTTP_CONTENT_TOO_LARGE = 413, HTTP_INTERNAL_SERVER_ERROR = 500, HTTP_SERVICE_UNAVAILABLE = 503, }; @@ -32,6 +33,7 @@ inline std::string_view HTTPStatusReasonString(HTTPStatusCode code) case HTTP_FORBIDDEN: return "Forbidden"; case HTTP_NOT_FOUND: return "Not Found"; case HTTP_BAD_METHOD: return "Method Not Allowed"; + case HTTP_CONTENT_TOO_LARGE: return "Content too large"; case HTTP_INTERNAL_SERVER_ERROR: return "Internal Server Error"; case HTTP_SERVICE_UNAVAILABLE: return "Service Unavailable"; } diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp index 0470e166974..c9b8eea5ffa 100644 --- a/src/test/httpserver_tests.cpp +++ b/src/test/httpserver_tests.cpp @@ -594,13 +594,18 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) BOOST_CHECK(actual.find("Content-Type: text/html; charset=ISO-8859-1\r\n") != std::string::npos); BOOST_CHECK(actual.find("Date: Wed, 11 Dec 2024 00:47:09 GMT\r\n") != std::string::npos); - // 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. + // Wait up to one minute for connection to be automatically closed, because + // keep-alive was not set by the client and we are done responding to their request. + attempts = 6000; + while (server.GetConnectionsCount() != 0) { + std::this_thread::sleep_for(10ms); + BOOST_REQUIRE(--attempts > 0); + } + + // Stop the I/O loop and shutdown server.InterruptNet(); + // Wait for I/O loop to finish, after all connected sockets are closed server.JoinSocketsThreads(); - // Close connection - BOOST_REQUIRE(server.CloseConnection(*client)); // Close all listening sockets server.StopListening(); }