Improve performance of Counters and add unit-test (#629)

Fixes #478

CMakeLists.txt

@@ -118,6 +118,7 @@ if(BUILD_TESTS)
     unittests/rpc/ErrorTests.cpp
     unittests/rpc/BaseTests.cpp
     unittests/rpc/RPCHelpersTest.cpp
+    unittests/rpc/CountersTest.cpp
     ## RPC handlers
     unittests/rpc/handlers/DefaultProcessorTests.cpp
     unittests/rpc/handlers/TestHandlerTests.cpp

src/rpc/Counters.cpp

@@ -18,33 +18,17 @@
 //==============================================================================
 
 #include <rpc/Counters.h>
+#include <rpc/JS.h>
 #include <rpc/RPC.h>
 #include <rpc/RPCHelpers.h>
 
 namespace RPC {
 
-void
-Counters::initializeCounter(std::string const& method)
-{
-    std::shared_lock lk(mutex_);
-    if (methodInfo_.count(method) == 0)
-    {
-        lk.unlock();
-        std::scoped_lock ulk(mutex_);
-
-        // This calls the default constructor for methodInfo of the method.
-        methodInfo_[method];
-    }
-}
-
 void
 Counters::rpcErrored(std::string const& method)
 {
-    initializeCounter(method);
+    std::scoped_lock lk(mutex_);
-
-    std::shared_lock lk(mutex_);
     MethodInfo& counters = methodInfo_[method];
-
     counters.started++;
     counters.errored++;
 }
@@ -52,11 +36,8 @@ Counters::rpcErrored(std::string const& method)
 void
 Counters::rpcComplete(std::string const& method, std::chrono::microseconds const& rpcDuration)
 {
-    initializeCounter(method);
+    std::scoped_lock lk(mutex_);
-
-    std::shared_lock lk(mutex_);
     MethodInfo& counters = methodInfo_[method];
-
     counters.started++;
     counters.finished++;
     counters.duration += rpcDuration.count();
@@ -65,18 +46,15 @@ Counters::rpcComplete(std::string const& method, std::chrono::microseconds const
 void
 Counters::rpcForwarded(std::string const& method)
 {
-    initializeCounter(method);
+    std::scoped_lock lk(mutex_);
-
-    std::shared_lock lk(mutex_);
     MethodInfo& counters = methodInfo_[method];
-
     counters.forwarded++;
 }
 
 boost::json::object
 Counters::report() const
 {
-    std::shared_lock lk(mutex_);
+    std::scoped_lock lk(mutex_);
     auto obj = boost::json::object{};
 
     obj[JS(rpc)] = boost::json::object{};

src/rpc/Counters.h

@@ -19,12 +19,12 @@
 
 #pragma once
 
-#include <boost/json.hpp>
-#include <chrono>
-#include <cstdint>
-#include <functional>
 #include <rpc/WorkQueue.h>
-#include <shared_mutex>
+
+#include <boost/json.hpp>
+
+#include <chrono>
+#include <mutex>
 #include <string>
 #include <unordered_map>
 
@@ -32,22 +32,16 @@ namespace RPC {
 
 class Counters
 {
-private:
     struct MethodInfo
     {
-        MethodInfo() = default;
+        std::uint64_t started = 0u;
-
+        std::uint64_t finished = 0u;
-        std::atomic_uint64_t started{0};
+        std::uint64_t errored = 0u;
-        std::atomic_uint64_t finished{0};
+        std::uint64_t forwarded = 0u;
-        std::atomic_uint64_t errored{0};
+        std::uint64_t duration = 0u;
-        std::atomic_uint64_t forwarded{0};
-        std::atomic_uint64_t duration{0};
     };
 
-    void
+    mutable std::mutex mutex_;
-    initializeCounter(std::string const& method);
-
-    mutable std::shared_mutex mutex_;
     std::unordered_map<std::string, MethodInfo> methodInfo_;
 
     std::reference_wrapper<const WorkQueue> workQueue_;

src/rpc/WorkQueue.cpp

@@ -27,3 +27,10 @@ WorkQueue::WorkQueue(std::uint32_t numWorkers, uint32_t maxSize)
     while (--numWorkers)
         threads_.emplace_back([this] { ioc_.run(); });
 }
+
+WorkQueue::~WorkQueue()
+{
+    work_.reset();
+    for (auto& thread : threads_)
+        thread.join();
+}

src/rpc/WorkQueue.h

@@ -48,6 +48,7 @@ class WorkQueue
 
 public:
     WorkQueue(std::uint32_t numWorkers, uint32_t maxSize = 0);
+    ~WorkQueue();
 
     static WorkQueue
     make_WorkQueue(clio::Config const& config)

unittests/rpc/BaseTests.cpp

@@ -2,9 +2,11 @@
 /*
     This file is part of clio: https://github.com/XRPLF/clio
     Copyright (c) 2023, 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

unittests/rpc/CountersTest.cpp

@@ -0,0 +1,67 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2023, 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.h>
+
+#include <rpc/Counters.h>
+#include <rpc/JS.h>
+
+#include <boost/json.hpp>
+#include <gtest/gtest.h>
+
+using namespace clio;
+using namespace RPC;
+
+class RPCCountersTest : public NoLoggerFixture
+{
+protected:
+    WorkQueue queue{4u, 1024u};  // todo: mock instead
+    Counters counters{queue};
+};
+
+TEST_F(RPCCountersTest, CheckThatCountersAddUp)
+{
+    for (auto i = 0u; i < 512; ++i)
+    {
+        counters.rpcErrored("error");
+        counters.rpcComplete("complete", std::chrono::milliseconds{1u});
+        counters.rpcForwarded("forward");
+    }
+
+    auto const report = counters.report();
+    auto const& rpc = report.at(JS(rpc)).as_object();
+
+    EXPECT_STREQ(rpc.at("error").as_object().at(JS(started)).as_string().c_str(), "512");
+    EXPECT_STREQ(rpc.at("error").as_object().at(JS(finished)).as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("error").as_object().at(JS(errored)).as_string().c_str(), "512");
+    EXPECT_STREQ(rpc.at("error").as_object().at("forwarded").as_string().c_str(), "0");
+
+    EXPECT_STREQ(rpc.at("complete").as_object().at(JS(started)).as_string().c_str(), "512");
+    EXPECT_STREQ(rpc.at("complete").as_object().at(JS(finished)).as_string().c_str(), "512");
+    EXPECT_STREQ(rpc.at("complete").as_object().at(JS(errored)).as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("complete").as_object().at("forwarded").as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("complete").as_object().at(JS(duration_us)).as_string().c_str(), "512000");  // 1000 per call
+
+    EXPECT_STREQ(rpc.at("forward").as_object().at(JS(started)).as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("forward").as_object().at(JS(finished)).as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("forward").as_object().at(JS(errored)).as_string().c_str(), "0");
+    EXPECT_STREQ(rpc.at("forward").as_object().at("forwarded").as_string().c_str(), "512");
+
+    EXPECT_EQ(report.at("work_queue"), queue.report());  // Counters report includes queue report
+}