diff --git a/src/common/restapi/RequestDispatcher.h b/src/common/restapi/RequestDispatcher.h index 84965203..41755060 100644 --- a/src/common/restapi/RequestDispatcher.h +++ b/src/common/restapi/RequestDispatcher.h @@ -518,6 +518,77 @@ namespace restapi reply = http::HTTPPayload::statusPayload(http::HTTPPayload::BAD_REQUEST, "application/json"); } + /** + * @brief Handles an HTTP request asynchronously on the handler executor. + * @tparam CompletionHandler Type invoked with the completed reply. + * @param request HTTP request. + * @param completionHandler Handler invoked with the completed reply. + */ + template + void handleRequestAsync(const Request& request, CompletionHandler completionHandler) + { + for (const auto& matcher : m_matchers) { + std::smatch what; + bool matched = false; + + if (!matcher.second->regex()) { + matched = request.uri.find(matcher.first) != std::string::npos; + } + else { + matched = std::regex_match(request.uri, what, std::regex(matcher.first)); + } + + if (!matched) + continue; + + const std::string expression = matcher.first; + MatcherTypePtr matchedHandler = matcher.second; + Request requestCopy = request; + std::smatch whatCopy = what; + const uint32_t warnSec = (m_warnSec > 0U) ? m_warnSec : 30U; + + m_executor->submit([matchedHandler, requestCopy, whatCopy, expression, warnSec, completionHandler]() mutable { + Reply workerReply; + const auto started = std::chrono::steady_clock::now(); + + workerReply.headers.add("Access-Control-Allow-Origin", "*"); + workerReply.headers.add("Access-Control-Allow-Methods", "*"); + workerReply.headers.add("Access-Control-Allow-Headers", "*"); + if (requestCopy.method == HTTP_OPTIONS) { + workerReply.status = http::HTTPPayload::OK; + } + + try { + matchedHandler->handleRequest(requestCopy, workerReply, whatCopy); + } + catch (const std::exception& e) { + ::LogError(LOG_REST, "REST handler exception, method = %s, uri = %s, expression = %s, error = %s", + requestCopy.method.c_str(), requestCopy.uri.c_str(), expression.c_str(), e.what()); + workerReply = http::HTTPPayload::statusPayload(http::HTTPPayload::INTERNAL_SERVER_ERROR, "application/json"); + } + catch (...) { + ::LogError(LOG_REST, "REST handler exception, method = %s, uri = %s, expression = %s", + requestCopy.method.c_str(), requestCopy.uri.c_str(), expression.c_str()); + workerReply = http::HTTPPayload::statusPayload(http::HTTPPayload::INTERNAL_SERVER_ERROR, "application/json"); + } + + const uint64_t elapsed = (uint64_t)std::chrono::duration_cast( + std::chrono::steady_clock::now() - started).count(); + if (elapsed >= warnSec) { + ::LogWarning(LOG_REST, "slow REST handler completed, method = %s, uri = %s, expression = %s, elapsed = %us", + requestCopy.method.c_str(), requestCopy.uri.c_str(), expression.c_str(), (uint32_t)elapsed); + } + + completionHandler(std::move(workerReply)); + }); + return; + } + + ::LogError(LOG_REST, "unknown endpoint, uri = %s", request.uri.c_str()); + Reply reply = http::HTTPPayload::statusPayload(http::HTTPPayload::BAD_REQUEST, "application/json"); + completionHandler(std::move(reply)); + } + private: typedef std::shared_ptr MatcherTypePtr; diff --git a/src/common/restapi/http/HTTPServer.h b/src/common/restapi/http/HTTPServer.h index 8c6c465f..e373dd2f 100644 --- a/src/common/restapi/http/HTTPServer.h +++ b/src/common/restapi/http/HTTPServer.h @@ -21,6 +21,7 @@ #include "common/restapi/http/HTTPRequestHandler.h" #include +#include #include #include #include @@ -60,6 +61,7 @@ namespace restapi m_acceptor(m_ioService), m_connectionManager(), m_socket(m_ioService), + m_acceptRetryTimer(m_ioService), m_requestHandler(), m_debug(debug) { @@ -124,6 +126,7 @@ namespace restapi // operations; once all operations have finished the m_ioService::run() // call will exit m_acceptor.close(); + m_acceptRetryTimer.cancel(); m_connectionManager.stopAll(); } @@ -140,11 +143,26 @@ namespace restapi return; } + if (ec == asio::error::operation_aborted) { + return; + } + if (!ec) { m_connectionManager.start(std::make_shared(std::move(m_socket), m_connectionManager, m_requestHandler, false, m_debug)); + accept(); + return; } - accept(); + ::LogError(LOG_REST, "HTTPServer::accept(), %s, code = %u; retrying", ec.message().c_str(), ec.value()); + asio::error_code ignoredError; + m_socket.close(ignoredError); + m_socket = asio::ip::tcp::socket(m_ioService); + m_acceptRetryTimer.expires_from_now(std::chrono::seconds(1U)); + m_acceptRetryTimer.async_wait([this](asio::error_code timerError) { + if (!timerError && m_acceptor.is_open()) { + accept(); + } + }); }); } @@ -159,6 +177,7 @@ namespace restapi ServerConnectionManager m_connectionManager; asio::ip::tcp::socket m_socket; + asio::steady_timer m_acceptRetryTimer; RequestHandlerType m_requestHandler; bool m_debug; diff --git a/src/common/restapi/http/SecureHTTPServer.h b/src/common/restapi/http/SecureHTTPServer.h index 25ff73d1..05e2aac6 100644 --- a/src/common/restapi/http/SecureHTTPServer.h +++ b/src/common/restapi/http/SecureHTTPServer.h @@ -23,6 +23,7 @@ #include "common/restapi/http/HTTPRequestHandler.h" #include +#include #include #include #include @@ -64,6 +65,7 @@ namespace restapi m_connectionManager(), m_context(asio::ssl::context::tlsv12), m_socket(m_ioService), + m_acceptRetryTimer(m_ioService), m_requestHandler(), m_debug(debug) { @@ -137,6 +139,7 @@ namespace restapi // operations; once all operations have finished the m_ioService::run() // call will exit m_acceptor.close(); + m_acceptRetryTimer.cancel(); m_connectionManager.stopAll(); } @@ -153,11 +156,26 @@ namespace restapi return; } + if (ec == asio::error::operation_aborted) { + return; + } + if (!ec) { m_connectionManager.start(std::make_shared(std::move(m_socket), m_context, m_connectionManager, m_requestHandler, false, m_debug)); + accept(); + return; } - accept(); + ::LogError(LOG_REST, "SecureHTTPServer::accept(), %s, code = %u; retrying", ec.message().c_str(), ec.value()); + asio::error_code ignoredError; + m_socket.close(ignoredError); + m_socket = asio::ip::tcp::socket(m_ioService); + m_acceptRetryTimer.expires_from_now(std::chrono::seconds(1U)); + m_acceptRetryTimer.async_wait([this](asio::error_code timerError) { + if (!timerError && m_acceptor.is_open()) { + accept(); + } + }); }); } @@ -173,6 +191,7 @@ namespace restapi asio::ssl::context m_context; asio::ip::tcp::socket m_socket; + asio::steady_timer m_acceptRetryTimer; std::string m_certFile; std::string m_keyFile; diff --git a/src/common/restapi/http/SecureServerConnection.h b/src/common/restapi/http/SecureServerConnection.h index 1c8b4397..433261da 100644 --- a/src/common/restapi/http/SecureServerConnection.h +++ b/src/common/restapi/http/SecureServerConnection.h @@ -23,6 +23,7 @@ #include "common/Log.h" #include +#include #include #include #include @@ -74,10 +75,13 @@ namespace restapi m_socket(std::move(socket), context), m_connectionManager(manager), m_requestHandler(handler), + m_deadline(m_socket.get_executor()), m_lexer(HTTPLexer(false)), m_continue(false), m_contResult(HTTPLexer::INDETERMINATE), m_headerBytes(0U), + m_stopped(false), + m_waitingForHandler(false), m_persistent(persistent), m_debug(debug) { @@ -87,12 +91,20 @@ namespace restapi /** * @brief Start the first asynchronous operation for the connection. */ - void start() { handshake(); } + void start() + { + armDeadline(HANDSHAKE_TIMEOUT_SECONDS, "TLS handshake"); + handshake(); + } /** * @brief Stop all asynchronous operations associated with the connection. */ void stop() { + m_stopped = true; + m_waitingForHandler = false; + asio::error_code ignoredError; + m_deadline.cancel(ignoredError); try { if (m_socket.lowest_layer().is_open()) { @@ -112,7 +124,14 @@ namespace restapi m_socket.async_handshake(asio::ssl::stream_base::server, [this, self](asio::error_code ec) { if (!ec) { + armDeadline(REQUEST_TIMEOUT_SECONDS, "request"); read(); + return; + } + + if (ec != asio::error::operation_aborted) { + ::LogError(LOG_REST, "SecureServerConnection::handshake(), %s, code = %u", ec.message().c_str(), ec.value()); + m_connectionManager.stop(self); } }); } @@ -195,13 +214,23 @@ namespace restapi m_continue = false; m_contResult = HTTPLexer::INDETERMINATE; m_headerBytes = 0U; - m_requestHandler.handleRequest(m_request, m_reply); - - if (m_debug) { - Utils::dump(1U, "SecureServerConnection::read(), HTTPS Reply Content", (uint8_t*)m_reply.content.c_str(), m_reply.content.length()); - } - - write(); + armDeadline(HANDLER_TIMEOUT_SECONDS, "handler"); + m_waitingForHandler = true; + dispatchRequest(m_requestHandler, m_request, + [this, self](HTTPPayload reply) mutable { + asio::post(m_socket.get_executor(), [this, self, reply]() mutable { + if (m_stopped || !m_waitingForHandler || !m_socket.lowest_layer().is_open()) + return; + + m_waitingForHandler = false; + cancelDeadline(); + m_reply = std::move(reply); + if (m_debug) { + Utils::dump(1U, "SecureServerConnection::read(), HTTPS Reply Content", (uint8_t*)m_reply.content.c_str(), m_reply.content.length()); + } + write(); + }); + }, 0); } else if (result == HTTPLexer::BAD) { m_continue = false; @@ -242,12 +271,15 @@ namespace restapi { auto self = this->shared_from_this(); + armDeadline(WRITE_TIMEOUT_SECONDS, "write"); + if (m_persistent) { m_reply.headers.add("Connection", "keep-alive"); } auto buffers = m_reply.toBuffers(); asio::async_write(m_socket, buffers, [this, self](asio::error_code ec, std::size_t) { + cancelDeadline(); if (m_persistent) { m_lexer.reset(); m_reply.headers = HTTPHeaders(); @@ -255,6 +287,7 @@ namespace restapi m_reply.content = ""; m_request = HTTPPayload(); m_headerBytes = 0U; + armDeadline(REQUEST_TIMEOUT_SECONDS, "request"); read(); } else { @@ -278,6 +311,72 @@ namespace restapi }); } + /** + * @brief Dispatches through an asynchronous request-handler interface when available. + * @tparam Handler Request-handler type. + * @tparam CompletionHandler Completion-handler type. + * @param handler Request handler. + * @param request HTTP request. + * @param completionHandler Handler invoked with the completed reply. + * @param priority Overload priority selector. + */ + template + static auto dispatchRequest(Handler& handler, const HTTPPayload& request, CompletionHandler completionHandler, int priority) + -> decltype(handler.handleRequestAsync(request, completionHandler), void()) + { + (void)priority; + handler.handleRequestAsync(request, completionHandler); + } + + /** + * @brief Dispatches through the legacy synchronous request-handler interface. + * @tparam Handler Request-handler type. + * @tparam CompletionHandler Completion-handler type. + * @param handler Request handler. + * @param request HTTP request. + * @param completionHandler Handler invoked with the completed reply. + */ + template + static void dispatchRequest(Handler& handler, const HTTPPayload& request, CompletionHandler completionHandler, long) + { + HTTPPayload reply; + handler.handleRequest(request, reply); + completionHandler(std::move(reply)); + } + + /** + * @brief Arms the connection deadline for the current operation. + * @param timeoutSeconds Deadline duration in seconds. + * @param operation Operation name used for diagnostics. + */ + void armDeadline(uint32_t timeoutSeconds, const char* operation) + { + auto self = this->shared_from_this(); + m_deadline.expires_from_now(std::chrono::seconds(timeoutSeconds)); + m_deadline.async_wait([this, self, operation](asio::error_code ec) { + if (!ec && !m_stopped) { + ::LogError(LOG_REST, "HTTPS connection timed out during %s", operation); + if (m_waitingForHandler) { + m_waitingForHandler = false; + m_reply = HTTPPayload::statusPayload(HTTPPayload::SERVICE_UNAVAILABLE, "application/json"); + write(); + } + else { + m_connectionManager.stop(self); + } + } + }); + } + + /** + * @brief Cancels the active connection deadline. + */ + void cancelDeadline() + { + asio::error_code ignoredError; + m_deadline.cancel(ignoredError); + } + /** * @brief Parses the Content-Length header value. * @param value The string value of the Content-Length header. @@ -310,6 +409,7 @@ namespace restapi ConnectionManagerType& m_connectionManager; RequestHandlerType& m_requestHandler; + asio::steady_timer m_deadline; std::array m_buffer; @@ -320,9 +420,15 @@ namespace restapi bool m_continue; HTTPLexer::ResultType m_contResult; size_t m_headerBytes; + bool m_stopped; + bool m_waitingForHandler; static constexpr size_t MAX_HTTP_HEADER_LENGTH = 32768U; static constexpr size_t MAX_HTTP_CONTENT_LENGTH = 1048576U; + static constexpr uint32_t HANDSHAKE_TIMEOUT_SECONDS = 15U; + static constexpr uint32_t REQUEST_TIMEOUT_SECONDS = 30U; + static constexpr uint32_t HANDLER_TIMEOUT_SECONDS = 60U; + static constexpr uint32_t WRITE_TIMEOUT_SECONDS = 30U; bool m_persistent; bool m_debug; diff --git a/src/common/restapi/http/ServerConnection.h b/src/common/restapi/http/ServerConnection.h index acd33b95..db97d325 100644 --- a/src/common/restapi/http/ServerConnection.h +++ b/src/common/restapi/http/ServerConnection.h @@ -22,6 +22,7 @@ #include "common/Utils.h" #include +#include #include #include #include @@ -71,10 +72,13 @@ namespace restapi m_socket(std::move(socket)), m_connectionManager(manager), m_requestHandler(handler), + m_deadline(m_socket.get_executor()), m_lexer(HTTPLexer(false)), m_continue(false), m_contResult(HTTPLexer::INDETERMINATE), m_headerBytes(0U), + m_stopped(false), + m_waitingForHandler(false), m_persistent(persistent), m_debug(debug) { @@ -84,12 +88,20 @@ namespace restapi /** * @brief Start the first asynchronous operation for the connection. */ - void start() { read(); } + void start() + { + armDeadline(REQUEST_TIMEOUT_SECONDS, "request"); + read(); + } /** * @brief Stop all asynchronous operations associated with the connection. */ void stop() { + m_stopped = true; + m_waitingForHandler = false; + asio::error_code ignoredError; + m_deadline.cancel(ignoredError); try { if (m_socket.is_open()) { @@ -179,13 +191,23 @@ namespace restapi m_continue = false; m_contResult = HTTPLexer::INDETERMINATE; m_headerBytes = 0U; - m_requestHandler.handleRequest(m_request, m_reply); - - if (m_debug) { - Utils::dump(1U, "ServerConnection::read(), HTTP Reply Content", (uint8_t*)m_reply.content.c_str(), m_reply.content.length()); - } - - write(); + armDeadline(HANDLER_TIMEOUT_SECONDS, "handler"); + m_waitingForHandler = true; + dispatchRequest(m_requestHandler, m_request, + [this, self](HTTPPayload reply) mutable { + asio::post(m_socket.get_executor(), [this, self, reply]() mutable { + if (m_stopped || !m_waitingForHandler || !m_socket.is_open()) + return; + + m_waitingForHandler = false; + cancelDeadline(); + m_reply = std::move(reply); + if (m_debug) { + Utils::dump(1U, "ServerConnection::read(), HTTP Reply Content", (uint8_t*)m_reply.content.c_str(), m_reply.content.length()); + } + write(); + }); + }, 0); } else if (result == HTTPLexer::BAD) { m_continue = false; @@ -227,12 +249,15 @@ namespace restapi { auto self = this->shared_from_this(); + armDeadline(WRITE_TIMEOUT_SECONDS, "write"); + if (m_persistent) { m_reply.headers.add("Connection", "keep-alive"); } auto buffers = m_reply.toBuffers(); asio::async_write(m_socket, buffers, [this, self](asio::error_code ec, std::size_t) { + cancelDeadline(); if (m_persistent) { m_lexer.reset(); m_reply.headers = HTTPHeaders(); @@ -240,6 +265,7 @@ namespace restapi m_reply.content = ""; m_request = HTTPPayload(); m_headerBytes = 0U; + armDeadline(REQUEST_TIMEOUT_SECONDS, "request"); read(); } else { @@ -263,6 +289,72 @@ namespace restapi }); } + /** + * @brief Dispatches through an asynchronous request-handler interface when available. + * @tparam Handler Request-handler type. + * @tparam CompletionHandler Completion-handler type. + * @param handler Request handler. + * @param request HTTP request. + * @param completionHandler Handler invoked with the completed reply. + * @param priority Overload priority selector. + */ + template + static auto dispatchRequest(Handler& handler, const HTTPPayload& request, CompletionHandler completionHandler, int priority) + -> decltype(handler.handleRequestAsync(request, completionHandler), void()) + { + (void)priority; + handler.handleRequestAsync(request, completionHandler); + } + + /** + * @brief Dispatches through the legacy synchronous request-handler interface. + * @tparam Handler Request-handler type. + * @tparam CompletionHandler Completion-handler type. + * @param handler Request handler. + * @param request HTTP request. + * @param completionHandler Handler invoked with the completed reply. + */ + template + static void dispatchRequest(Handler& handler, const HTTPPayload& request, CompletionHandler completionHandler, long) + { + HTTPPayload reply; + handler.handleRequest(request, reply); + completionHandler(std::move(reply)); + } + + /** + * @brief Arms the connection deadline for the current operation. + * @param timeoutSeconds Deadline duration in seconds. + * @param operation Operation name used for diagnostics. + */ + void armDeadline(uint32_t timeoutSeconds, const char* operation) + { + auto self = this->shared_from_this(); + m_deadline.expires_from_now(std::chrono::seconds(timeoutSeconds)); + m_deadline.async_wait([this, self, operation](asio::error_code ec) { + if (!ec && !m_stopped) { + ::LogError(LOG_REST, "HTTP connection timed out during %s", operation); + if (m_waitingForHandler) { + m_waitingForHandler = false; + m_reply = HTTPPayload::statusPayload(HTTPPayload::SERVICE_UNAVAILABLE, "application/json"); + write(); + } + else { + m_connectionManager.stop(self); + } + } + }); + } + + /** + * @brief Cancels the active connection deadline. + */ + void cancelDeadline() + { + asio::error_code ignoredError; + m_deadline.cancel(ignoredError); + } + /** * @brief Parses the Content-Length header value. * @param value The string value of the Content-Length header. @@ -295,6 +387,7 @@ namespace restapi ConnectionManagerType& m_connectionManager; RequestHandlerType& m_requestHandler; + asio::steady_timer m_deadline; std::array m_buffer; @@ -305,9 +398,14 @@ namespace restapi bool m_continue; HTTPLexer::ResultType m_contResult; size_t m_headerBytes; + bool m_stopped; + bool m_waitingForHandler; static constexpr size_t MAX_HTTP_HEADER_LENGTH = 32768U; static constexpr size_t MAX_HTTP_CONTENT_LENGTH = 1048576U; + static constexpr uint32_t REQUEST_TIMEOUT_SECONDS = 30U; + static constexpr uint32_t HANDLER_TIMEOUT_SECONDS = 60U; + static constexpr uint32_t WRITE_TIMEOUT_SECONDS = 30U; bool m_persistent; bool m_debug; diff --git a/tests/network/HTTP_RestAPI_Tests.cpp b/tests/network/HTTP_RestAPI_Tests.cpp index f0a7c865..502bc5c2 100644 --- a/tests/network/HTTP_RestAPI_Tests.cpp +++ b/tests/network/HTTP_RestAPI_Tests.cpp @@ -14,6 +14,7 @@ #include "common/restapi/http/HTTPClient.h" #include "common/restapi/http/HTTPPayload.h" #include "common/restapi/http/HTTPServer.h" +#include "common/restapi/RequestDispatcher.h" #include #include @@ -81,6 +82,35 @@ namespace { return response; } + /** + * @brief Sends a raw HTTP request without using test assertions in the calling thread. + * @param port The port to which the HTTP request should be sent. + * @param request Complete HTTP request. + * @return The raw HTTP response received from the server. + */ + std::string sendRawRequestUnchecked(uint16_t port, const std::string& request) + { + asio::io_service ioService; + asio::ip::tcp::socket socket(ioService); + asio::error_code ec; + socket.connect(asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), port), ec); + if (ec) + return std::string(); + + asio::write(socket, asio::buffer(request), ec); + if (ec) + return std::string(); + + std::string response; + std::array buffer {}; + do { + const size_t length = socket.read_some(asio::buffer(buffer), ec); + response.append(buffer.data(), length); + } while (!ec); + + return response; + } + std::string responseBody(const std::string& response) { const size_t separator = response.find("\r\n\r\n"); @@ -402,3 +432,50 @@ TEST_CASE("HTTP server rejects bytes beyond the declared body", "[restapi][http] REQUIRE(response.find("HTTP/1.0 400 Bad Request\r\n") == 0U); REQUIRE(fixture.state->calls == 0U); } + +TEST_CASE("HTTP server services another connection while a REST handler is running", "[restapi][http][dispatcher][e2e]") +{ + std::atomic slowStarted { false }; + restapi::DefaultRequestDispatcher dispatcher; + dispatcher.match("/slow").get([&slowStarted](const HTTPPayload&, HTTPPayload& reply, const restapi::RequestMatch&) { + slowStarted = true; + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + std::string content = "slow"; + reply.payload(content, HTTPPayload::OK, "text/plain"); + }); + dispatcher.match("/fast").get([](const HTTPPayload&, HTTPPayload& reply, const restapi::RequestMatch&) { + std::string content = "fast"; + reply.payload(content, HTTPPayload::OK, "text/plain"); + }); + + HTTPServer server("127.0.0.1", 0U, false); + server.setHandler(dispatcher); + server.open(); + const uint16_t port = server.localPort(); + std::thread serverThread([&server]() { server.run(); }); + + const std::string slowRequest = "GET /slow HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n"; + const std::string fastRequest = "GET /fast HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n"; + std::string slowResponse; + std::thread slowThread([&]() { slowResponse = sendRawRequestUnchecked(port, slowRequest); }); + + const auto startDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!slowStarted && std::chrono::steady_clock::now() < startDeadline) + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + const bool slowObserved = slowStarted; + + const auto fastStarted = std::chrono::steady_clock::now(); + const std::string fastResponse = sendRawRequestUnchecked(port, fastRequest); + const auto fastElapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - fastStarted); + + slowThread.join(); + server.stop(); + serverThread.join(); + + REQUIRE(slowObserved); + REQUIRE(fastResponse.find("HTTP/1.0 200 OK\r\n") == 0U); + REQUIRE(responseBody(fastResponse) == "fast"); + REQUIRE(fastElapsed < std::chrono::milliseconds(400)); + REQUIRE(responseBody(slowResponse) == "slow"); +}