From cdf71998e554ed844c3e1e878bba25c090a7875e Mon Sep 17 00:00:00 2001 From: Matthew Zipkin Date: Tue, 10 Dec 2024 20:02:55 -0500 Subject: [PATCH] HTTPServer: compose and send replies to connected clients Sockets-touching bits copied and adapted from `CConnman::SocketSendData()` Testing this requires adding a new feature to the SocketTestingSetup, returning the DynSock I/O pipes from the mock socket so the received data can be checked. Co-authored-by: Vasil Dimov --- src/httpserver.cpp | 183 +++++++++++++++++++++++++++++++-- src/httpserver.h | 31 ++++++ src/test/httpserver_tests.cpp | 38 ++++++- src/test/util/setup_common.cpp | 4 +- src/test/util/setup_common.h | 7 +- 5 files changed, 251 insertions(+), 12 deletions(-) diff --git a/src/httpserver.cpp b/src/httpserver.cpp index dffe791581b..572fa140258 100644 --- a/src/httpserver.cpp +++ b/src/httpserver.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #include #include @@ -967,6 +968,101 @@ bool HTTPRequest::LoadBody(LineReader& reader) } } +void HTTPRequest::WriteReply(HTTPStatusCode status, std::span reply_body) +{ + HTTPResponse res; + + // Some response headers are determined in advance and stored in the request + res.m_headers = std::move(m_response_headers); + + // Response version matches request version + res.m_version = m_version; + + // Add response code + res.m_status = status; + + // See libevent evhttp_response_needs_body() + // Response headers are different if no body is needed + 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}; + + // See libevent evhttp_make_header_response() + // Expected response headers depend on protocol version + if (m_version.major == 1) { + // HTTP/1.0 + if (m_version.minor == 0) { + auto connection_header{m_headers.FindFirst("Connection")}; + if (connection_header && ToLower(connection_header.value()) == "keep-alive") { + res.m_headers.Write("Connection", "keep-alive"); + keep_alive = true; + // HTTP/1.0 connections are closed by default so EOF is sufficient + // to indicate end of the body. Adding Content-Length a special case. + if (needs_body) needs_content_length = true; + } + } + + // HTTP/1.1 + if (m_version.minor >= 1) { + const int64_t now_seconds{TicksSinceEpoch(NodeClock::now())}; + res.m_headers.Write("Date", FormatRFC1123DateTime(now_seconds)); + + // HTTP/1.1 connections are kept alive by default and always require Content-Length. + if (needs_body) needs_content_length = true; + + // Default for HTTP/1.1 + keep_alive = true; + } + } + + if (needs_content_length) { + res.m_headers.Write("Content-Length", util::ToString(reply_body.size())); + } + + if (needs_body && !res.m_headers.FindFirst("Content-Type")) { + // Default type from libevent evhttp_new_object() + res.m_headers.Write("Content-Type", "text/html; charset=ISO-8859-1"); + } + + auto connection_header{m_headers.FindFirst("Connection")}; + if (connection_header && ToLower(connection_header.value()) == "close") { + // Might not exist already but we need to replace it, not append to it + res.m_headers.RemoveAll("Connection"); + + res.m_headers.Write("Connection", "close"); + keep_alive = false; + } + + // Serialize the response headers + const std::string headers{res.StringifyHeaders()}; + const auto headers_bytes{std::as_bytes(std::span{headers})}; + + // Fill the send buffer with the complete serialized response headers + body + { + LOCK(m_client->m_send_mutex); + m_client->m_send_buffer.insert(m_client->m_send_buffer.end(), headers_bytes.begin(), headers_bytes.end()); + + // We've been using std::span up until now but it is finally time to copy + // data. The original data will go out of scope when WriteReply() returns. + // This is analogous to the memcpy() in libevent's evbuffer_add() + m_client->m_send_buffer.insert(m_client->m_send_buffer.end(), reply_body.begin(), reply_body.end()); + } + + // Inform HTTPServer I/O loop that there is data that is ready to be sent to + // this client in the next loop iteration. + m_client->m_send_ready = true; + + LogDebug( + BCLog::HTTP, + "HTTPResponse (status code: %d size: %lld) added to send buffer for client %s (id=%llu)", + status, + headers_bytes.size() + reply_body.size(), + m_client->m_origin, + m_client->m_id); +} + util::Expected HTTPServer::BindAndStartListening(const CService& to) { // Create socket for listening for incoming connections @@ -1156,7 +1252,12 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const bool err_ready = events.occurred & Sock::ERR; if (send_ready) { - // TODO: send data + // Try to send as much data as is ready for this client. + // If there's an error we can skip the receive phase for this client + // because we need to disconnect. + if (!client->MaybeSendBytesFromBuffer()) { + recv_ready = false; + } } if (recv_ready || err_ready) { @@ -1225,15 +1326,12 @@ HTTPServer::IOReadiness HTTPServer::GenerateWaitSockets() const } 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); + // Check if client is ready to send data. Don't try to receive again + // until the send buffer is cleared (all data sent to client). + Sock::Event event = (http_client->m_send_ready ? Sock::SEND : Sock::RECV); io_readiness.events_per_sock.emplace(sock, Sock::Events{event}); io_readiness.httpclients_per_sock.emplace(sock, http_client); } @@ -1318,4 +1416,75 @@ bool HTTPRemoteClient::ReadRequest(HTTPRequest& req) return true; } + +bool HTTPRemoteClient::MaybeSendBytesFromBuffer() +{ + // Send as much data from this client's buffer as we can + LOCK(m_send_mutex); + if (!m_send_buffer.empty()) { + // Socket flags (See kernel docs for send(2) and tcp(7) for more details). + // MSG_NOSIGNAL: If the remote end of the connection is closed, + // fail with EPIPE (an error) as opposed to triggering + // SIGPIPE which terminates the process. + // MSG_DONTWAIT: Makes the send operation non-blocking regardless of socket blocking mode. + // MSG_MORE: We do not set this flag here because http responses are usually + // small and we want the kernel to send them right away. Setting MSG_MORE + // would "cork" the socket to prevent sending out partial frames. + int flags{MSG_NOSIGNAL | MSG_DONTWAIT}; + + // Try to send bytes through socket + ssize_t bytes_sent; + { + LOCK(m_sock_mutex); + bytes_sent = m_sock->Send(m_send_buffer.data(), + m_send_buffer.size(), + flags); + } + + if (bytes_sent < 0) { + // Something went wrong + const int err{WSAGetLastError()}; + if (!IOErrorIsPermanent(err)) { + // The error can be safely ignored, try the send again on the next I/O loop. + m_send_ready = true; + return true; + } else { + // Unrecoverable error, log and disconnect client. + LogDebug( + BCLog::HTTP, + "Error sending HTTP response data to client %s (id=%llu): %s", + m_origin, + m_id, + NetworkErrorString(err)); + // TODO: disconnect + // Do not attempt to read from this client. + return false; + } + } + + // Successful send, remove sent bytes from our local buffer. + Assume(static_cast(bytes_sent) <= m_send_buffer.size()); + m_send_buffer.erase(m_send_buffer.begin(), + m_send_buffer.begin() + bytes_sent); + + LogDebug( + BCLog::HTTP, + "Sent %d bytes to client %s (id=%llu)", + bytes_sent, + m_origin, + m_id); + + // This check is inside the if(!empty) block meaning "there was data but now its gone". + // We wouldn't want to change the flags if MaybeSendBytesFromBuffer() was called + // 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; + } + } + + return true; +} } // namespace http_bitcoin diff --git a/src/httpserver.h b/src/httpserver.h index 01a97922924..26345c0fdf8 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -295,6 +295,9 @@ public: //! Pointer to the client that made the request so we know who to respond to. std::shared_ptr m_client; + //! Response headers may be set in advance before response body is known + HTTPHeaders m_response_headers; + explicit HTTPRequest(std::shared_ptr client) : m_client{std::move(client)} {} //! Construct with a null client for unit tests explicit HTTPRequest() : m_client{} {} @@ -312,6 +315,12 @@ public: bool LoadHeaders(LineReader& reader); bool LoadBody(LineReader& reader); /// @} + + void WriteReply(HTTPStatusCode status, std::span reply_body = {}); + void WriteReply(HTTPStatusCode status, std::string_view reply_body_view) + { + WriteReply(status, std::as_bytes(std::span{reply_body_view})); + } }; class HTTPServer @@ -518,6 +527,22 @@ public: */ std::vector m_recv_buffer{}; + /** + * Response data destined for this client. + * Written to by http worker threads, read and erased by HTTPServer I/O thread + */ + /// @{ + Mutex m_send_mutex; + std::vector m_send_buffer GUARDED_BY(m_send_mutex); + /// @} + + /** + * Set true by worker threads after writing a response to m_send_buffer. + * Set false by the HTTPServer I/O thread after flushing m_send_buffer. + * Checked in the HTTPServer I/O loop to avoid locking m_send_mutex if there's nothing to send. + */ + std::atomic_bool m_send_ready{false}; + /** * 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 @@ -547,6 +572,12 @@ public: * @returns true upon reading a complete request, otherwise false (may throw). */ bool ReadRequest(HTTPRequest& req); + + /** + * Push data (if there is any) from client's m_send_buffer to the connected socket. + * @returns false if we are done with this client and HTTPServer can skip the next read operation from it. + */ + bool MaybeSendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex); }; } // namespace http_bitcoin diff --git a/src/test/httpserver_tests.cpp b/src/test/httpserver_tests.cpp index 1c90cdc29d0..0470e166974 100644 --- a/src/test/httpserver_tests.cpp +++ b/src/test/httpserver_tests.cpp @@ -489,6 +489,10 @@ BOOST_AUTO_TEST_CASE(http_request_tests) BOOST_AUTO_TEST_CASE(http_server_socket_tests) { + // Hard code the timestamp for the Date header in the HTTP response + // Wed Dec 11 00:47:09 2024 UTC + SetMockTime(1733878029); + // Prepare a request handler that just stores received requests so we can examine them. // Mutex is required to prevent a race between this test's main thread and the server's I/O loop. Mutex requests_mutex; @@ -524,8 +528,10 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) // No connections yet BOOST_CHECK_EQUAL(server.GetConnectionsCount(), 0); - // Create a mock client with pre-loaded request data and add it to the local CreateSock queue - ConnectClient(std::as_bytes(std::span(full_request))); + // Create a mock client with pre-loaded request data and add it to the local CreateSock queue. + // Keep a handle for the mock client's send and receive pipes so we can examine + // the data it "receives". + std::shared_ptr mock_client_socket_pipes{ConnectClient(std::as_bytes(std::span(full_request)))}; // Wait up to a minute to find and connect the client in the I/O loop int attempts{6000}; @@ -553,6 +559,9 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) client = requests.front()->m_client; BOOST_CHECK_EQUAL(client->m_origin, "5.5.5.5:6789"); + // Respond to request + requests.front()->WriteReply(HTTP_OK, "874140\n"); + break; } } @@ -560,6 +569,31 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests) BOOST_REQUIRE(--attempts > 0); } + // Check the sent response from the mock client at the other end of the mock socket + std::string actual; + // Wait up to one minute for all the bytes to appear in the "send" pipe. + char buf[0x10000] = {}; + attempts = 6000; + while (attempts > 0) + { + ssize_t bytes_read = mock_client_socket_pipes->send.GetBytes(buf, sizeof(buf), 0); + if (bytes_read > 0) { + actual.append(buf, bytes_read); + if (actual.length() == 146) { + break; + } + } + std::this_thread::sleep_for(10ms); + --attempts; + } + BOOST_CHECK(actual.starts_with("HTTP/1.1 200 OK\r\n")); + BOOST_CHECK(actual.ends_with("\r\n874140\n")); + // Headers can be sorted in any order, and will be, since we use unordered_map + BOOST_CHECK(actual.find("Connection: close\r\n") != std::string::npos); + BOOST_CHECK(actual.find("Content-Length: 7\r\n") != std::string::npos); + 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. diff --git a/src/test/util/setup_common.cpp b/src/test/util/setup_common.cpp index 6a4f8ff178b..9e82eae8e66 100644 --- a/src/test/util/setup_common.cpp +++ b/src/test/util/setup_common.cpp @@ -658,7 +658,7 @@ SocketTestingSetup::~SocketTestingSetup() CreateSock = m_create_sock_orig; } -void SocketTestingSetup::ConnectClient(std::span data) +std::shared_ptr SocketTestingSetup::ConnectClient(std::span data) { // I/O pipes for a mock Connected Socket we can read and write to. auto connected_socket_pipes(std::make_shared()); @@ -672,6 +672,8 @@ void SocketTestingSetup::ConnectClient(std::span data) // Push into the queue of Accepted Sockets returned by the local CreateSock() m_accepted_sockets.Push(std::move(connected_socket)); + + return connected_socket_pipes; } /** diff --git a/src/test/util/setup_common.h b/src/test/util/setup_common.h index 6773a689b84..e44c7e728bf 100644 --- a/src/test/util/setup_common.h +++ b/src/test/util/setup_common.h @@ -256,8 +256,11 @@ public: explicit SocketTestingSetup(); ~SocketTestingSetup(); - //! Connect to the socket with a mock client (a DynSock) and send pre-loaded data. - void ConnectClient(std::span data); + /** + * Connect to the socket with a mock client (a DynSock) and send pre-loaded data. + * Returns the I/O pipes from the mock client so we can read response data sent to it. + */ + std::shared_ptr ConnectClient(std::span data); private: //! Save the original value of CreateSock here and restore it when the test ends.