tighten up how the shared maps and vectors lock themselves during certain operations; implement some critical changes to the REST API to help prevent and log on stuck REST API calls; correct some lock order issues within the FNE REST API;

1 month ago · 55bef73a5b
parent 45c0661723
commit 55bef73a5b
4 changed files with 317 additions and 19 deletions
--- 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;
        }

        /**
--- 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;
        }

        /**
--- 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 <string>
 #include <regex>
 #include <memory>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstdint>
+#include <deque>
+#include <future>
+#include <mutex>
+#include <thread>
+#include <vector>

 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<bool> 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::milliseconds>(
+            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<std::chrono::milliseconds>(deadline - now).count();
+    }
+
+    // ---------------------------------------------------------------------------
+    //  Structure Declaration
+    // ---------------------------------------------------------------------------
+
    /**
     * @brief Structure representing a REST API request match.
     * @ingroup rest
@ -159,22 +250,122 @@ namespace restapi
    template<typename Request = http::HTTPPayload, typename Reply = http::HTTPPayload>
    class RequestDispatcher {
        typedef RequestMatcher<Request, Reply> 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<void()> task;
+
+                            {
+                                std::unique_lock<std::mutex> 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<std::mutex> 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<void> Future representing the completion of the task.
+             */
+            template<typename Callable>
+            std::future<void> submit(Callable&& callable)
+            {
+                std::packaged_task<void()> task(std::forward<Callable>(callable));
+                std::future<void> future = task.get_future();
+
+                {
+                    std::lock_guard<std::mutex> lock(m_mutex);
+                    m_queue.emplace_back(std::move(task));
+                }
+
+                m_cv.notify_one();
+                return future;
+            }
+
+        private:
+            std::vector<std::thread> m_workers;
+            std::deque<std::packaged_task<void()>> 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<HandlerExecutor>()) { /* 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<HandlerExecutor>()) { /* 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<HandlerExecutor>()) { /* 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<RequestExecutionContext>(request.method, request.uri, expression, timeoutSec);
+                auto workerReply = std::make_shared<Reply>();
+
+                Request requestCopy = request;
+                std::smatch whatCopy = what;
+
+                std::future<void> 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<void> 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::seconds>(
+                        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<void> monitorFut = sharedFut;
+                        std::shared_ptr<RequestExecutionContext> 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::seconds>(
+                                    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::seconds>(
+                                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()) {
@ -226,7 +499,7 @@ namespace restapi
                            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<std::string, MatcherTypePtr> m_matchers;

        bool m_debug;
+        uint32_t m_warnSec;
+        uint32_t m_timeoutSec;
+        std::shared_ptr<HandlerExecutor> m_executor;
    };

    // ---------------------------------------------------------------------------
--- 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<uint32_t>(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<json::array>(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<uint32_t> peerIds = std::vector<uint32_t>();

@ -2176,6 +2179,7 @@ void RESTAPI::restAPI_GetAffList(const HTTPPayload& request, HTTPPayload& reply,
                }
            }
        }
+        m_network->m_peers.shared_unlock();

        response["totalAffiliations"].set<uint32_t>(totalAffiliations);
    }