stats: use scheduler for queue processing instead of thread

Author: kwvg

src/init.cpp

@@ -2302,6 +2302,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
         int nStatsPeriod = std::min(std::max((int)args.GetIntArg("-statsperiod", DEFAULT_STATSD_PERIOD), MIN_STATSD_PERIOD), MAX_STATSD_PERIOD);
         node.scheduler->scheduleEvery(std::bind(&PeriodicStats, std::ref(node)), std::chrono::seconds{nStatsPeriod});
     }
+    ::g_stats_client->Schedule(*node.scheduler);
 
     // ********************************************************* Step 11: import blocks
 

src/stats/client.cpp

@@ -10,6 +10,7 @@
 #include <random.h>
 #include <stats/rawsender.h>
 #include <sync.h>
+#include <util/check.h>
 #include <util/strencodings.h>
 #include <util/system.h>
 #include <util/translation.h>
@@ -49,6 +50,8 @@ class StatsdClientImpl final : public StatsdClient
     ~StatsdClientImpl() = default;
 
 public:
+    void Schedule(CScheduler& scheduler) override;
+
     bool dec(std::string_view key, float sample_rate) override { return count(key, -1, sample_rate); }
     bool inc(std::string_view key, float sample_rate) override { return count(key, 1, sample_rate); }
     bool count(std::string_view key, int64_t delta, float sample_rate) override { return _send(key, delta, STATSD_METRIC_COUNT, sample_rate); }
@@ -185,6 +188,11 @@ StatsdClientImpl::StatsdClientImpl(const std::string& host, uint16_t port, bool
     LogPrintf("%s: Initialized to transmit stats to %s:%d\n", __func__, host, port);
 }
 
+void StatsdClientImpl::Schedule(CScheduler& scheduler)
+{
+    Assert(m_sender)->Schedule(scheduler);
+}
+
 template <typename T1>
 inline bool StatsdClientImpl::_send(std::string_view key, T1 value, std::string_view type, float sample_rate)
 {

src/stats/client.h

@@ -15,6 +15,7 @@
 #include <string_view>
 
 class ArgsManager;
+class CScheduler;
 
 /** Default host assumed to be running a Statsd server */
 static const std::string DEFAULT_STATSD_HOST{""};
@@ -40,6 +41,9 @@ class StatsdClient
     static util::Result<std::unique_ptr<StatsdClient>> make(const ArgsManager& args);
     virtual ~StatsdClient() = default;
 
+    /* Schedule tasks */
+    virtual void Schedule(CScheduler& scheduler) {}
+
     /* Statsd-defined APIs */
     virtual bool dec(std::string_view key, float sample_rate = 1.f) { return false; }
     virtual bool inc(std::string_view key, float sample_rate = 1.f) { return false; }

src/stats/rawsender.cpp

@@ -8,6 +8,7 @@
 #include <logging.h>
 #include <netaddress.h>
 #include <netbase.h>
+#include <scheduler.h>
 #include <util/sock.h>
 #include <util/thread.h>
 
@@ -38,26 +39,19 @@ RawSender::RawSender(const std::string& host, uint16_t port, bool use_tcp, std::
 
     if (m_interval_ms == 0) {
         LogPrint(BCLog::NET, "%s: Send interval is zero, not starting RawSender queueing thread.\n", __func__);
-    } else {
-        m_interrupt.reset();
-        m_thread = std::thread(&util::TraceThread, "rawsender", [this] { QueueThreadMain(); });
     }
 
     if (m_use_tcp) {
         m_reconn = std::thread(&util::TraceThread, "rawreconnect", [this] { ReconnectThread(); });
     }
 
-    LogPrint(BCLog::NET, "%s: Started %sinstance sending messages to %s over %s\n", __func__,
-             m_thread.joinable() ? "threaded " : "", this->ToStringHostPort(), m_use_tcp ? "TCP" : "UDP");
+    LogPrint(BCLog::NET, "%s: Started instance sending messages to %s over %s\n", __func__, this->ToStringHostPort(),
+             m_use_tcp ? "TCP" : "UDP");
 }
 
 RawSender::~RawSender()
 {
     // If there are threads, interrupt and stop it
-    if (m_thread.joinable()) {
-        m_interrupt();
-        m_thread.join();
-    }
     if (m_reconn.joinable()) {
         m_reconn_interrupt();
         m_reconn.join();
@@ -67,6 +61,12 @@ RawSender::~RawSender()
     QueueFlush(m_queue);
 }
 
+void RawSender::Schedule(CScheduler& scheduler)
+{
+    if (m_interval_ms == 0) return;
+    scheduler.scheduleEvery([this] { this->QueueThreadMain(); }, std::chrono::milliseconds{m_interval_ms});
+}
+
 std::optional<bilingual_str> RawSender::Connect()
 {
     AssertLockHeld(cs_net);
@@ -167,8 +167,8 @@ std::optional<bilingual_str> RawSender::Send(const RawMessage& msg)
     AssertLockNotHeld(cs);
     AssertLockNotHeld(cs_net);
 
-    // If there is a thread, append to queue
-    if (m_thread.joinable()) {
+    // If queueing is enabled, append
+    if (m_interval_ms > 0) {
         WITH_LOCK(cs, QueueAdd(m_queue, msg));
         return std::nullopt;
     }
@@ -283,14 +283,8 @@ void RawSender::QueueThreadMain()
     AssertLockNotHeld(cs);
     AssertLockNotHeld(cs_net);
 
-    while (!m_interrupt) {
-        // Swap the queues to commit the existing queue of messages
-        std::vector<RawMessage> queue;
-        WITH_LOCK(cs, m_queue.swap(queue));
-        QueueFlush(queue);
-
-        if (!m_interrupt.sleep_for(std::chrono::milliseconds(m_interval_ms))) {
-            return;
-        }
-    }
+    // Swap the queues to commit the existing queue of messages
+    std::vector<RawMessage> queue;
+    WITH_LOCK(cs, m_queue.swap(queue));
+    QueueFlush(queue);
 }

src/stats/rawsender.h

@@ -17,6 +17,7 @@
 #include <string>
 #include <vector>
 
+class CScheduler;
 class Sock;
 
 struct RawMessage : public std::vector<uint8_t>
@@ -64,6 +65,8 @@ class RawSender
     RawSender& operator=(const RawSender&) = delete;
     RawSender(RawSender&&) = delete;
 
+    void Schedule(CScheduler& schedule);
+
     //! Request a message to be sent based on configuration (queueing, batching)
     std::optional<bilingual_str> Send(const RawMessage& msg) EXCLUSIVE_LOCKS_REQUIRED(!cs, !cs_net);
 
@@ -111,14 +114,10 @@ class RawSender
 
     /* Mutex to protect (batches of) messages queue */
     mutable Mutex cs;
-    /* Interrupt for queue processing thread */
-    CThreadInterrupt m_interrupt;
     /* Interrupt for reconnection thread (TCP only) */
     CThreadInterrupt m_reconn_interrupt;
     /* Queue of (batches of) messages to be sent */
     std::vector<RawMessage> m_queue GUARDED_BY(cs);
-    /* Thread that processes queue every m_interval_ms */
-    std::thread m_thread;
     /* Reconnection attempt thread (TCP only) */
     std::thread m_reconn;
     /* Queue of messages to be sent when reconnection succeeds (TCP only) */