diff --git a/src/common/concurrent/shared_unordered_map.h b/src/common/concurrent/shared_unordered_map.h index bbcaa8b9..f87c80ff 100644 --- a/src/common/concurrent/shared_unordered_map.h +++ b/src/common/concurrent/shared_unordered_map.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2025 Bryan Biedenkapp, N2PLL + * Copyright (C) 2025,2026 Bryan Biedenkapp, N2PLL * */ /** @@ -311,7 +311,10 @@ namespace concurrent */ iterator find(const Key& key) { - return m_map.find(key); + __shared_lock(); + iterator it = m_map.find(key); + __shared_unlock(); + return it; } /** * @brief Tries to locate an element in an unordered_map. @@ -321,7 +324,10 @@ namespace concurrent */ const_iterator find(const Key& key) const { - return m_map.find(key); + __shared_lock(); + const_iterator it = m_map.find(key); + __shared_unlock(); + return it; } /** diff --git a/src/common/concurrent/shared_vector.h b/src/common/concurrent/shared_vector.h index 34aac3ac..ae571300 100644 --- a/src/common/concurrent/shared_vector.h +++ b/src/common/concurrent/shared_vector.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2025 Bryan Biedenkapp, N2PLL + * Copyright (C) 2025,2026 Bryan Biedenkapp, N2PLL * */ /** @@ -265,7 +265,10 @@ namespace concurrent */ T& front() { - return m_vector.front(); + __shared_lock(); + T& value = m_vector.front(); + __shared_unlock(); + return value; } /** * @brief Gets the first element of the vector. @@ -273,7 +276,10 @@ namespace concurrent */ const T& front() const { - return m_vector.front(); + __shared_lock(); + const T& value = m_vector.front(); + __shared_unlock(); + return value; } /** @@ -282,7 +288,10 @@ namespace concurrent */ T& back() { - return m_vector.back(); + __shared_lock(); + T& value = m_vector.back(); + __shared_unlock(); + return value; } /** * @brief Gets the last element of the vector. @@ -290,7 +299,10 @@ namespace concurrent */ const T& back() const { - return m_vector.back(); + __shared_lock(); + const T& value = m_vector.back(); + __shared_unlock(); + return value; } /** diff --git a/src/common/restapi/RequestDispatcher.h b/src/common/restapi/RequestDispatcher.h index 1470be32..84965203 100644 --- a/src/common/restapi/RequestDispatcher.h +++ b/src/common/restapi/RequestDispatcher.h @@ -4,7 +4,7 @@ * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * - * Copyright (C) 2023 Bryan Biedenkapp, N2PLL + * Copyright (C) 2023,2026 Bryan Biedenkapp, N2PLL * */ /** @@ -29,6 +29,15 @@ #include #include #include +#include +#include +#include +#include +#include +#include +#include +#include +#include namespace restapi { @@ -36,6 +45,88 @@ namespace restapi // Structure Declaration // --------------------------------------------------------------------------- + /** + * @brief Per-request cooperative cancellation context. + * @ingroup rest + */ + struct RequestExecutionContext { + RequestExecutionContext(std::string m, std::string u, std::string e, uint32_t timeoutSec) : + cancelled(false), + started(std::chrono::steady_clock::now()), + timeout(std::chrono::seconds(timeoutSec)), + method(std::move(m)), + uri(std::move(u)), + expression(std::move(e)) + { + /* stub */ + } + + std::atomic cancelled; + std::chrono::steady_clock::time_point started; + std::chrono::steady_clock::duration timeout; + std::string method; + std::string uri; + std::string expression; + }; + + /** + * @brief Internal helper to access request context TLS. + */ + inline const RequestExecutionContext*& requestContextTLS() + { + static thread_local const RequestExecutionContext* s_ctx = nullptr; + return s_ctx; + } + + /** + * @brief Returns true when current request has been cooperatively cancelled. + * @ingroup rest + */ + inline bool currentRequestCancelled() + { + const RequestExecutionContext* ctx = requestContextTLS(); + return (ctx != nullptr) ? ctx->cancelled.load() : false; + } + + /** + * @brief Returns elapsed milliseconds for the current request, or 0 if none. + * @ingroup rest + */ + inline uint64_t currentRequestElapsedMs() + { + const RequestExecutionContext* ctx = requestContextTLS(); + if (ctx == nullptr) { + return 0U; + } + + return (uint64_t)std::chrono::duration_cast( + std::chrono::steady_clock::now() - ctx->started).count(); + } + + /** + * @brief Returns remaining milliseconds before request timeout, or 0 if none/expired. + * @ingroup rest + */ + inline uint64_t currentRequestRemainingMs() + { + const RequestExecutionContext* ctx = requestContextTLS(); + if (ctx == nullptr) { + return 0U; + } + + auto now = std::chrono::steady_clock::now(); + auto deadline = ctx->started + ctx->timeout; + if (now >= deadline) { + return 0U; + } + + return (uint64_t)std::chrono::duration_cast(deadline - now).count(); + } + + // --------------------------------------------------------------------------- + // Structure Declaration + // --------------------------------------------------------------------------- + /** * @brief Structure representing a REST API request match. * @ingroup rest @@ -159,22 +250,122 @@ namespace restapi template class RequestDispatcher { typedef RequestMatcher MatcherType; + + // --------------------------------------------------------------------------- + // Class Declaration + // --------------------------------------------------------------------------- + + /** + * @brief Fixed-size executor for dispatching request handlers with deadlines. + */ + class HandlerExecutor { + public: + /** + * @brief Initializes a new instance of the HandlerExecutor class. + * @param workers Number of worker threads to use. If 0, defaults to number of hardware threads (capped at 8). + */ + explicit HandlerExecutor(size_t workers = 0U) : + m_workers(), + m_queue(), + m_mutex(), + m_cv(), + m_stop(false) + { + size_t count = workers; + if (count == 0U) { + count = std::thread::hardware_concurrency(); + if (count == 0U) { + count = 4U; + } + if (count > 8U) { + count = 8U; + } + } + + for (size_t i = 0U; i < count; i++) { + m_workers.emplace_back([this]() { + while (true) { + std::packaged_task task; + + { + std::unique_lock lock(m_mutex); + m_cv.wait(lock, [this]() { return m_stop || !m_queue.empty(); }); + if (m_stop && m_queue.empty()) { + return; + } + + task = std::move(m_queue.front()); + m_queue.pop_front(); + } + + task(); + } + }); + } + } + /** + * @brief Finalizes a instance of the HandlerExecutor class. + */ + ~HandlerExecutor() + { + { + std::lock_guard lock(m_mutex); + m_stop = true; + } + m_cv.notify_all(); + + for (auto& worker : m_workers) { + if (worker.joinable()) { + worker.join(); + } + } + } + + /** + * @brief Submits a callable task for execution. + * @tparam Callable Type of the callable task. + * @param callable Callable task to execute. + * @returns std::future Future representing the completion of the task. + */ + template + std::future submit(Callable&& callable) + { + std::packaged_task task(std::forward(callable)); + std::future future = task.get_future(); + + { + std::lock_guard lock(m_mutex); + m_queue.emplace_back(std::move(task)); + } + + m_cv.notify_one(); + return future; + } + + private: + std::vector m_workers; + std::deque> m_queue; + std::mutex m_mutex; + std::condition_variable m_cv; + bool m_stop; + }; + public: /** * @brief Initializes a new instance of the RequestDispatcher class. */ - RequestDispatcher() : m_basePath(), m_debug(false) { /* stub */ } + RequestDispatcher() : m_basePath(), m_debug(false), m_warnSec(30U), m_timeoutSec(60U), m_executor(std::make_shared()) { /* stub */ } /** * @brief Initializes a new instance of the RequestDispatcher class. * @param debug Flag indicating whether or not verbose logging should be enabled. */ - RequestDispatcher(bool debug) : m_basePath(), m_debug(debug) { /* stub */ } + RequestDispatcher(bool debug) : m_basePath(), m_debug(debug), m_warnSec(30U), m_timeoutSec(60U), m_executor(std::make_shared()) { /* stub */ } /** * @brief Initializes a new instance of the RequestDispatcher class. * @param basePath * @param debug Flag indicating whether or not verbose logging should be enabled. */ - RequestDispatcher(const std::string& basePath, bool debug) : m_basePath(basePath), m_debug(debug) { /* stub */ } + RequestDispatcher(const std::string& basePath, bool debug) : m_basePath(basePath), m_debug(debug), m_warnSec(30U), m_timeoutSec(60U), m_executor(std::make_shared()) { /* stub */ } /** * @brief Helper to match a request patch. @@ -207,6 +398,88 @@ namespace restapi */ void handleRequest(const Request& request, Reply& reply) { + // dispatch matched handlers on a bounded worker pool to support cooperative + // cancellation and timeout response without changing endpoint signatures. + auto guardedDispatch = [&](const std::string& expression, MatcherType& matcher, const std::smatch& what) { + const uint32_t warnSec = (m_warnSec > 0U) ? m_warnSec : 30U; + const uint32_t timeoutSec = (m_timeoutSec > warnSec) ? m_timeoutSec : 60U; + + auto requestCtx = std::make_shared(request.method, request.uri, expression, timeoutSec); + auto workerReply = std::make_shared(); + + Request requestCopy = request; + std::smatch whatCopy = what; + + std::future fut = m_executor->submit([&matcher, requestCopy, whatCopy, requestCtx, workerReply]() { + const RequestExecutionContext* prevCtx = requestContextTLS(); + requestContextTLS() = requestCtx.get(); + + try { + matcher.handleRequest(requestCopy, *workerReply, whatCopy); + } + catch (...) { + requestContextTLS() = prevCtx; + throw; + } + + requestContextTLS() = prevCtx; + }); + + std::shared_future sharedFut = fut.share(); + + bool warned = false; + bool timedOut = false; + auto nextStillLog = std::chrono::steady_clock::now() + std::chrono::seconds(timeoutSec); + + while (sharedFut.wait_for(std::chrono::seconds(1U)) != std::future_status::ready) { + const uint64_t elapsed = (uint64_t)std::chrono::duration_cast( + std::chrono::steady_clock::now() - requestCtx->started).count(); + + if (!warned && elapsed >= warnSec) { + warned = true; + ::LogWarning(LOG_REST, "slow REST handler, method = %s, uri = %s, expression = %s, elapsed = %us", + request.method.c_str(), request.uri.c_str(), expression.c_str(), (uint32_t)elapsed); + } + + if (!timedOut && elapsed >= timeoutSec) { + timedOut = true; + requestCtx->cancelled.store(true); + + ::LogError(LOG_REST, "stalled REST handler, method = %s, uri = %s, expression = %s, elapsed = %us (cooperative cancel requested)", + request.method.c_str(), request.uri.c_str(), expression.c_str(), (uint32_t)elapsed); + + // keep watcher active in a detached monitor to report completion after cancel. + std::shared_future monitorFut = sharedFut; + std::shared_ptr monitorCtx = requestCtx; + std::thread([monitorFut, monitorCtx, timeoutSec]() { + auto stillLogEvery = std::chrono::seconds(timeoutSec); + while (monitorFut.wait_for(stillLogEvery) != std::future_status::ready) { + uint64_t monitorElapsed = (uint64_t)std::chrono::duration_cast( + std::chrono::steady_clock::now() - monitorCtx->started).count(); + ::LogError(LOG_REST, "REST handler still running after cancel, method = %s, uri = %s, expression = %s, elapsed = %us", + monitorCtx->method.c_str(), monitorCtx->uri.c_str(), monitorCtx->expression.c_str(), (uint32_t)monitorElapsed); + } + + uint64_t totalElapsed = (uint64_t)std::chrono::duration_cast( + std::chrono::steady_clock::now() - monitorCtx->started).count(); + ::LogWarning(LOG_REST, "REST handler completed after cancellation request, method = %s, uri = %s, expression = %s, elapsed = %us", + monitorCtx->method.c_str(), monitorCtx->uri.c_str(), monitorCtx->expression.c_str(), (uint32_t)totalElapsed); + }).detach(); + + reply = http::HTTPPayload::statusPayload(http::HTTPPayload::SERVICE_UNAVAILABLE, "application/json"); + return; + } + + if (timedOut && std::chrono::steady_clock::now() >= nextStillLog) { + nextStillLog += std::chrono::seconds(timeoutSec); + } + } + + // worker completed before timeout; propagate any exception and then reply. + sharedFut.get(); + reply = *workerReply; + }; + for (const auto& matcher : m_matchers) { std::smatch what; if (!matcher.second->regex()) { @@ -225,8 +498,8 @@ namespace restapi if (request.method == HTTP_OPTIONS) { reply.status = http::HTTPPayload::OK; } - - matcher.second->handleRequest(request, reply, what); + + guardedDispatch(matcher.first, *matcher.second, what); return; } } else { @@ -235,7 +508,7 @@ namespace restapi ::LogDebug(LOG_REST, "regex endpoint, uri = %s, expression = %s", request.uri.c_str(), matcher.first.c_str()); } - matcher.second->handleRequest(request, reply, what); + guardedDispatch(matcher.first, *matcher.second, what); return; } } @@ -252,6 +525,9 @@ namespace restapi std::map m_matchers; bool m_debug; + uint32_t m_warnSec; + uint32_t m_timeoutSec; + std::shared_ptr m_executor; }; // --------------------------------------------------------------------------- diff --git a/src/fne/restapi/RESTAPI.cpp b/src/fne/restapi/RESTAPI.cpp index b1419b75..216ee033 100644 --- a/src/fne/restapi/RESTAPI.cpp +++ b/src/fne/restapi/RESTAPI.cpp @@ -875,8 +875,8 @@ void RESTAPI::restAPI_GetPeerQuery(const HTTPPayload& request, HTTPPayload& repl json::array peers = json::array(); if (m_network != nullptr) { + m_network->m_peers.shared_lock(); if (m_network->m_peers.size() > 0) { - m_network->m_peers.shared_lock(); for (auto entry : m_network->m_peers) { uint32_t peerId = entry.first; network::FNEPeerConnection* peer = entry.second; @@ -889,11 +889,11 @@ void RESTAPI::restAPI_GetPeerQuery(const HTTPPayload& request, HTTPPayload& repl peers.push_back(json::value(peerObj)); } } - m_network->m_peers.shared_unlock(); } else { LogError(LOG_REST, "peer query failed, no peers connected to this FNE"); } + m_network->m_peers.shared_unlock(); // report any peers from replica peers if (m_network->m_peerReplicaPeers.size() > 0) { @@ -930,7 +930,9 @@ void RESTAPI::restAPI_GetPeerCount(const HTTPPayload& request, HTTPPayload& repl json::array peers = json::array(); if (m_network != nullptr) { + m_network->m_peers.shared_lock(); uint32_t count = m_network->m_peers.size(); + m_network->m_peers.shared_unlock(); response["peerCount"].set(count); } @@ -1854,8 +1856,8 @@ void RESTAPI::restAPI_GetStats(const HTTPPayload& request, HTTPPayload& reply, c if (m_network != nullptr) { // peer statistics (right now this is just a list of connected peers) json::array peerStats = json::array(); + m_network->m_peers.shared_lock(); if (m_network->m_peers.size() > 0) { - m_network->m_peers.shared_lock(); for (auto entry : m_network->m_peers) { uint32_t peerId = entry.first; network::FNEPeerConnection* peer = entry.second; @@ -1905,8 +1907,8 @@ void RESTAPI::restAPI_GetStats(const HTTPPayload& request, HTTPPayload& reply, c peerStats.push_back(json::value(peerObj)); } } - m_network->m_peers.shared_unlock(); } + m_network->m_peers.shared_unlock(); response["peerStats"].set(peerStats); // table load statistics @@ -2136,6 +2138,7 @@ void RESTAPI::restAPI_GetAffList(const HTTPPayload& request, HTTPPayload& reply, json::array affs = json::array(); if (m_network != nullptr) { uint32_t totalAffiliations = 0U; + m_network->m_peers.shared_lock(); if (m_network->m_peers.size() > 0) { std::vector peerIds = std::vector(); @@ -2176,6 +2179,7 @@ void RESTAPI::restAPI_GetAffList(const HTTPPayload& request, HTTPPayload& reply, } } } + m_network->m_peers.shared_unlock(); response["totalAffiliations"].set(totalAffiliations); }