Add signals handler (#1365)

For #442.

docs/configure-clio.md

@@ -97,3 +97,11 @@ To enable the caching for a source, `forwarding_cache_timeout` value should be a
 
 `forwarding_cache_timeout` defines for how long (in seconds) a cache entry will be valid after being placed into the cache.
 Zero value turns off the cache feature.
+
+## Graceful shutdown (not fully implemented yet)
+
+Clio can be gracefully shut down by sending a `SIGINT` (Ctrl+C) or `SIGTERM` signal.
+The process will stop accepting new connections and will wait for the time specified in `graceful_period` config value (or 10 seconds by default).
+If Clio finishes all the scheduled operations before the end of the period, it will stop immediately.
+Otherwise, it will wait for the period to finish and then exit without finishing operations.
+If Clio receives a second signal during the period, it will stop immediately.

docs/examples/config/example-config.json

@@ -66,6 +66,8 @@
         // It's true by default unless admin_password is set,'local_admin' : true and 'admin_password' can not be set at the same time
         "local_amdin": false
     },
+    // Time in seconds for graceful shutdown. Defaults to 10 seconds. Not fully implemented yet.
+    "graceful_period": 10.,
     // Overrides log level on a per logging channel.
     // Defaults to global "log_level" for each unspecified channel.
     "log_channels": [

src/etl/LoadBalancer.cpp

@@ -24,6 +24,7 @@
 #include "etl/ETLService.hpp"
 #include "etl/ETLState.hpp"
 #include "etl/Source.hpp"
+#include "util/Constants.hpp"
 #include "util/Random.hpp"
 #include "util/log/Logger.hpp"
 
@@ -36,6 +37,7 @@
 
 #include <algorithm>
 #include <chrono>
+#include <cmath>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
@@ -72,9 +74,9 @@ LoadBalancer::LoadBalancer(
 {
     auto const forwardingCacheTimeout = config.valueOr<float>("forwarding_cache_timeout", 0.f);
     if (forwardingCacheTimeout > 0.f) {
-        forwardingCache_ = impl::ForwardingCache{
-            std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::duration<float>{forwardingCacheTimeout})
-        };
+        forwardingCache_ = impl::ForwardingCache{std::chrono::milliseconds{
+            std::lroundf(forwardingCacheTimeout * static_cast<float>(util::MILLISECONDS_PER_SECOND))
+        }};
     }
 
     static constexpr std::uint32_t MAX_DOWNLOAD = 256;

src/util/CMakeLists.txt

@@ -13,6 +13,7 @@ target_sources(
           prometheus/Prometheus.cpp
           Random.cpp
           Retry.cpp
+          SignalsHandler.cpp
           requests/RequestBuilder.cpp
           requests/Types.cpp
           requests/WsConnection.cpp

src/util/SignalsHandler.cpp

@@ -0,0 +1,140 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "util/SignalsHandler.hpp"
+
+#include "util/Assert.hpp"
+#include "util/Constants.hpp"
+#include "util/config/Config.hpp"
+#include "util/log/Logger.hpp"
+
+#include <boost/asio/error.hpp>
+#include <boost/asio/executor_work_guard.hpp>
+
+#include <chrono>
+#include <cmath>
+#include <csignal>
+#include <cstddef>
+#include <functional>
+#include <optional>
+#include <string>
+#include <utility>
+
+namespace util {
+
+namespace impl {
+
+class SignalsHandlerStatic {
+    static SignalsHandler* handler_;
+
+public:
+    static void
+    registerHandler(SignalsHandler& handler)
+    {
+        ASSERT(handler_ == nullptr, "There could be only one instance of SignalsHandler");
+        handler_ = &handler;
+    }
+
+    static void
+    resetHandler()
+    {
+        handler_ = nullptr;
+    }
+
+    static void
+    handleSignal(int signal)
+    {
+        ASSERT(handler_ != nullptr, "SignalsHandler is not initialized");
+        handler_->stopHandler_(signal);
+    }
+
+    static void
+    handleSecondSignal(int signal)
+    {
+        ASSERT(handler_ != nullptr, "SignalsHandler is not initialized");
+        handler_->secondSignalHandler_(signal);
+    }
+};
+
+SignalsHandler* SignalsHandlerStatic::handler_ = nullptr;
+
+}  // namespace impl
+
+SignalsHandler::SignalsHandler(Config const& config, std::function<void()> forceExitHandler)
+    : gracefulPeriod_(0)
+    , context_(1)
+    , stopHandler_([this, forceExitHandler](int) mutable {
+        LOG(LogService::info()) << "Got stop signal. Stopping Clio. Graceful period is "
+                                << std::chrono::duration_cast<std::chrono::milliseconds>(gracefulPeriod_).count()
+                                << " milliseconds.";
+        setHandler(impl::SignalsHandlerStatic::handleSecondSignal);
+        timer_.emplace(context_.scheduleAfter(
+            gracefulPeriod_,
+            [forceExitHandler = std::move(forceExitHandler)](auto&& stopToken, bool canceled) {
+                // TODO: Update this after https://github.com/XRPLF/clio/issues/1367
+                if (not stopToken.isStopRequested() and not canceled) {
+                    LOG(LogService::warn()) << "Force exit at the end of graceful period.";
+                    forceExitHandler();
+                }
+            }
+        ));
+        stopSignal_();
+    })
+    , secondSignalHandler_([this, forceExitHandler = std::move(forceExitHandler)](int) {
+        LOG(LogService::warn()) << "Force exit on second signal.";
+        forceExitHandler();
+        cancelTimer();
+        setHandler();
+    })
+{
+    impl::SignalsHandlerStatic::registerHandler(*this);
+
+    auto const gracefulPeriod =
+        std::round(config.valueOr("graceful_period", 10.f) * static_cast<float>(util::MILLISECONDS_PER_SECOND));
+    ASSERT(gracefulPeriod >= 0.f, "Graceful period must be non-negative");
+    gracefulPeriod_ = std::chrono::milliseconds{static_cast<size_t>(gracefulPeriod)};
+
+    setHandler(impl::SignalsHandlerStatic::handleSignal);
+}
+
+SignalsHandler::~SignalsHandler()
+{
+    cancelTimer();
+    setHandler();
+    impl::SignalsHandlerStatic::resetHandler();  // This is needed mostly for tests to reset static state
+}
+
+void
+SignalsHandler::cancelTimer()
+{
+    if (timer_.has_value()) {
+        timer_->cancel();
+        timer_->requestStop();
+    }
+}
+
+void
+SignalsHandler::setHandler(void (*handler)(int))
+{
+    for (int const signal : HANDLED_SIGNALS) {
+        std::signal(signal, handler == nullptr ? SIG_DFL : handler);
+    }
+}
+
+}  // namespace util

src/util/SignalsHandler.hpp

@@ -0,0 +1,121 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#pragma once
+
+#include "util/async/context/BasicExecutionContext.hpp"
+#include "util/config/Config.hpp"
+#include "util/log/Logger.hpp"
+
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <boost/signals2/signal.hpp>
+#include <boost/signals2/variadic_signal.hpp>
+
+#include <chrono>
+#include <concepts>
+#include <csignal>
+#include <cstdlib>
+#include <functional>
+#include <optional>
+#include <string>
+
+namespace util {
+
+namespace impl {
+class SignalsHandlerStatic;
+}  // namespace impl
+
+/**
+ * @brief Class handling signals.
+ * @note There could be only one instance of this class.
+ */
+class SignalsHandler {
+    std::chrono::steady_clock::duration gracefulPeriod_;
+    async::PoolExecutionContext context_;
+    std::optional<async::PoolExecutionContext::ScheduledOperation<void>> timer_;
+
+    boost::signals2::signal<void()> stopSignal_;
+    std::function<void(int)> stopHandler_;
+    std::function<void(int)> secondSignalHandler_;
+
+    friend class impl::SignalsHandlerStatic;
+
+public:
+    /**
+     * @brief Enum for stop priority.
+     */
+    enum class Priority { StopFirst = 0, Normal = 1, StopLast = 2 };
+
+    /**
+     * @brief Create SignalsHandler object.
+     *
+     * @param config The configuration.
+     * @param forceExitHandler The handler for forced exit.
+     */
+    SignalsHandler(Config const& config, std::function<void()> forceExitHandler = defaultForceExitHandler_);
+
+    SignalsHandler(SignalsHandler const&) = delete;
+    SignalsHandler(SignalsHandler&&) = delete;
+    SignalsHandler&
+    operator=(SignalsHandler const&) = delete;
+    SignalsHandler&
+    operator=(SignalsHandler&&) = delete;
+
+    /**
+     * @brief Destructor of SignalsHandler.
+     */
+    ~SignalsHandler();
+
+    /**
+     * @brief Subscribe to stop signal.
+     *
+     * @tparam SomeCallback The type of the callback.
+     * @param callback The callback to call on stop signal.
+     * @param priority The priority of the callback. Default is Normal.
+     */
+    template <std::invocable SomeCallback>
+    void
+    subscribeToStop(SomeCallback&& callback, Priority priority = Priority::Normal)
+    {
+        stopSignal_.connect(static_cast<int>(priority), std::forward<SomeCallback>(callback));
+    }
+
+    static constexpr auto HANDLED_SIGNALS = {SIGINT, SIGTERM};
+
+private:
+    /**
+     * @brief Cancel scheduled force exit if any.
+     */
+    void
+    cancelTimer();
+
+    /**
+     * @brief Set signal handler for handled signals.
+     *
+     * @param handler The handler to set. Default is nullptr.
+     */
+    static void
+    setHandler(void (*handler)(int) = nullptr);
+
+    static auto constexpr defaultForceExitHandler_ = []() { std::exit(EXIT_FAILURE); };
+};
+
+}  // namespace util

unittests/CMakeLists.txt

@@ -8,13 +8,13 @@ target_sources(
           data/BackendFactoryTests.cpp
           data/BackendInterfaceTests.cpp
           data/cassandra/AsyncExecutorTests.cpp
-          # Webserver
           data/cassandra/BackendTests.cpp
           data/cassandra/BaseTests.cpp
           data/cassandra/ExecutionStrategyTests.cpp
           data/cassandra/RetryPolicyTests.cpp
           data/cassandra/SettingsProviderTests.cpp
           DOSGuardTests.cpp
+          # ETL
           etl/AmendmentBlockHandlerTests.cpp
           etl/CacheLoaderSettingsTests.cpp
           etl/CacheLoaderTests.cpp
@@ -32,7 +32,7 @@ target_sources(
           etl/SubscriptionSourceDependenciesTests.cpp
           etl/SubscriptionSourceTests.cpp
           etl/TransformerTests.cpp
-          # RPC
+          # Feed
           feed/BookChangesFeedTests.cpp
           feed/ForwardFeedTests.cpp
           feed/LedgerFeedTests.cpp
@@ -46,6 +46,7 @@ target_sources(
           Main.cpp
           Playground.cpp
           ProfilerTests.cpp
+          # RPC
           rpc/AmendmentsTests.cpp
           rpc/APIVersionTests.cpp
           rpc/BaseTests.cpp
@@ -90,16 +91,15 @@ target_sources(
           rpc/handlers/UnsubscribeTests.cpp
           rpc/handlers/VersionHandlerTests.cpp
           rpc/JsonBoolTests.cpp
-          # RPC handlers
           rpc/RPCHelpersTests.cpp
           rpc/WorkQueueTests.cpp
           util/AssertTests.cpp
+          # Async framework
           util/async/AnyExecutionContextTests.cpp
           util/async/AnyOperationTests.cpp
           util/async/AnyStopTokenTests.cpp
           util/async/AnyStrandTests.cpp
           util/async/AsyncExecutionContextTests.cpp
-          # Requests framework
           util/BatchingTests.cpp
           util/LedgerUtilsTests.cpp
           # Prometheus support
@@ -112,23 +112,23 @@ target_sources(
           util/prometheus/MetricBuilderTests.cpp
           util/prometheus/MetricsFamilyTests.cpp
           util/prometheus/OStreamTests.cpp
+          # Requests framework
           util/requests/RequestBuilderTests.cpp
           util/requests/SslContextTests.cpp
           util/requests/WsConnectionTests.cpp
-          # ETL
           util/RetryTests.cpp
-          # Async framework
+          util/SignalsHandlerTests.cpp
           util/StringUtils.cpp
           util/TestGlobals.cpp
           util/TestHttpServer.cpp
           util/TestObject.cpp
           util/TestWsServer.cpp
           util/TxUtilTests.cpp
+          # Webserver
           web/AdminVerificationTests.cpp
           web/RPCServerHandlerTests.cpp
           web/ServerTests.cpp
           web/SweepHandlerTests.cpp
-          # Feed
           web/WhitelistHandlerTests.cpp
 )
 

unittests/util/SignalsHandlerTests.cpp

@@ -0,0 +1,163 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "util/Fixtures.hpp"
+#include "util/SignalsHandler.hpp"
+#include "util/config/Config.hpp"
+
+#include <boost/json/value.hpp>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <condition_variable>
+#include <csignal>
+#include <cstddef>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <thread>
+
+using namespace util;
+using testing::MockFunction;
+using testing::StrictMock;
+
+struct SignalsHandlerTestsBase : NoLoggerFixture {
+    StrictMock<MockFunction<void()>> forceExitHandler_;
+    StrictMock<MockFunction<void()>> stopHandler_;
+    StrictMock<MockFunction<void()>> anotherStopHandler_;
+
+    void
+    allowTestToFinish()
+    {
+        std::unique_lock lock{mutex_};
+        testCanBeFinished_ = true;
+        cv_.notify_one();
+    }
+
+    void
+    wait()
+    {
+        std::unique_lock lock{mutex_};
+        cv_.wait(lock, [this] { return testCanBeFinished_; });
+    }
+
+    std::mutex mutex_;
+    std::condition_variable cv_;
+    bool testCanBeFinished_{false};
+};
+
+TEST(SignalsHandlerDeathTest, CantCreateTwoSignalsHandlers)
+{
+    auto makeHandler = []() { return SignalsHandler{Config{}, []() {}}; };
+    auto const handler = makeHandler();
+    EXPECT_DEATH({ makeHandler(); }, ".*");
+}
+
+struct SignalsHandlerTests : SignalsHandlerTestsBase {
+    SignalsHandler handler_{
+        util::Config{boost::json::value{{"graceful_period", 3.0}}},
+        forceExitHandler_.AsStdFunction()
+    };
+};
+
+TEST_F(SignalsHandlerTests, NoSignal)
+{
+    handler_.subscribeToStop(stopHandler_.AsStdFunction());
+    handler_.subscribeToStop(anotherStopHandler_.AsStdFunction());
+}
+
+TEST_F(SignalsHandlerTests, OneSignal)
+{
+    handler_.subscribeToStop(stopHandler_.AsStdFunction());
+    handler_.subscribeToStop(anotherStopHandler_.AsStdFunction());
+    EXPECT_CALL(stopHandler_, Call());
+    EXPECT_CALL(anotherStopHandler_, Call()).WillOnce([this]() { allowTestToFinish(); });
+    std::raise(SIGINT);
+
+    wait();
+}
+
+struct SignalsHandlerTimeoutTests : SignalsHandlerTestsBase {
+    SignalsHandler handler_{
+        util::Config{boost::json::value{{"graceful_period", 0.001}}},
+        forceExitHandler_.AsStdFunction()
+    };
+};
+
+TEST_F(SignalsHandlerTimeoutTests, OneSignalTimeout)
+{
+    handler_.subscribeToStop(stopHandler_.AsStdFunction());
+    EXPECT_CALL(stopHandler_, Call()).WillOnce([] { std::this_thread::sleep_for(std::chrono::milliseconds(2)); });
+    EXPECT_CALL(forceExitHandler_, Call());
+    std::raise(SIGINT);
+}
+
+TEST_F(SignalsHandlerTests, TwoSignals)
+{
+    handler_.subscribeToStop(stopHandler_.AsStdFunction());
+    EXPECT_CALL(stopHandler_, Call()).WillOnce([] { std::raise(SIGINT); });
+    EXPECT_CALL(forceExitHandler_, Call()).WillOnce([this]() { allowTestToFinish(); });
+    std::raise(SIGINT);
+
+    wait();
+}
+
+struct SignalsHandlerPriorityTestsBundle {
+    std::string name;
+    SignalsHandler::Priority stopHandlerPriority;
+    SignalsHandler::Priority anotherStopHandlerPriority;
+};
+
+struct SignalsHandlerPriorityTests : SignalsHandlerTests,
+                                     testing::WithParamInterface<SignalsHandlerPriorityTestsBundle> {};
+
+INSTANTIATE_TEST_SUITE_P(
+    SignalsHandlerPriorityTestsGroup,
+    SignalsHandlerPriorityTests,
+    testing::Values(
+        SignalsHandlerPriorityTestsBundle{
+            "StopFirst-Normal",
+            SignalsHandler::Priority::StopFirst,
+            SignalsHandler::Priority::Normal
+        },
+        SignalsHandlerPriorityTestsBundle{
+            "Normal-StopLast",
+            SignalsHandler::Priority::Normal,
+            SignalsHandler::Priority::StopLast
+        }
+    )
+);
+
+TEST_P(SignalsHandlerPriorityTests, Priority)
+{
+    bool stopHandlerCalled = false;
+
+    handler_.subscribeToStop(anotherStopHandler_.AsStdFunction(), GetParam().anotherStopHandlerPriority);
+    handler_.subscribeToStop(stopHandler_.AsStdFunction(), GetParam().stopHandlerPriority);
+
+    EXPECT_CALL(stopHandler_, Call()).WillOnce([&] { stopHandlerCalled = true; });
+    EXPECT_CALL(anotherStopHandler_, Call()).WillOnce([&] {
+        EXPECT_TRUE(stopHandlerCalled);
+        allowTestToFinish();
+    });
+    std::raise(SIGINT);
+
+    wait();
+}