Add throttler

CMakeLists.txt

@@ -16,6 +16,8 @@ set(HEADERS
     src/App.h
     src/Console.h
     src/Cpu.h
+    src/Monitor.h
+    src/Throttler.h
     src/interfaces/IClientListener.h
     src/interfaces/IConsoleListener.h
     src/interfaces/IJobResultListener.h
@@ -73,6 +75,8 @@ set(SOURCES
     src/log/FileLog.cpp
     src/log/Log.cpp
     src/Mem.cpp
+    src/Monitor.cpp
+    src/Throttler.cpp
     src/net/Client.cpp
     src/net/Job.cpp
     src/net/Network.cpp

build.sh

@@ -0,0 +1,8 @@
+#!/bin/sh
+set -vx
+rm -rf ./build
+mkdir -p ./build
+cd build
+cmake .. -DWITH_LIBCPUID=OFF -DWITH_AEON=OFF -DWITH_HTTPD=OFF
+make
+

src/App.cpp

@@ -41,6 +41,8 @@
 #include "Summary.h"
 #include "version.h"
 #include "workers/Workers.h"
+#include "Monitor.h"
+#include "Throttler.h"
 
 
 #ifdef HAVE_SYSLOG_H
@@ -65,6 +67,8 @@ App::App(int argc, char **argv) :
     m_self = this;
 
     Cpu::init();
+    Monitor::init(40); // TODO: parameterize
+    Throttler::init(1); // TODO
     m_options = Options::parse(argc, argv);
     if (!m_options) {
         return;

src/Monitor.cpp

@@ -0,0 +1,80 @@
+#include "Monitor.h"
+#include <iostream>
+#include <sys/resource.h>
+#include <sys/time.h>
+
+uint64_t Monitor::threshold;
+int Monitor::index;
+uv_timer_t Monitor::m_timer;
+time_t Monitor::timestamps[BUF_SIZE];
+uint64_t Monitor::percentages[BUF_SIZE];
+struct rusage Monitor::prevUsage;
+
+void Monitor::init(const uint64_t theThreshold)
+{
+    index = 0;
+    threshold = theThreshold;
+
+    for (int i = 0; i < BUF_SIZE; i++) {
+        timestamps[i] = 0;
+        percentages[i] = 0;
+    }
+    uv_timer_init(uv_default_loop(), &m_timer);
+    uv_timer_start(&m_timer, Monitor::onAddRecord, INITIAL_WAIT_S * 1000, INTERVAL_S * 1000);
+}
+
+uint64_t Monitor::getCurrentPercentage()
+{
+    uint64_t percentage = 0;
+    struct rusage usage;
+    getrusage(RUSAGE_SELF, &usage);
+    if (prevUsage.ru_utime.tv_sec != 0) {
+        time_t deltaSecond = (usage.ru_utime.tv_sec - prevUsage.ru_utime.tv_sec) + (usage.ru_stime.tv_sec - prevUsage.ru_stime.tv_sec);
+        suseconds_t deltaMicroSecond = (usage.ru_utime.tv_usec - prevUsage.ru_utime.tv_usec) + (usage.ru_stime.tv_usec - prevUsage.ru_stime.tv_usec);
+        percentage = (deltaSecond * 1000000 + deltaMicroSecond) / (INTERVAL_S * 10000);
+    }
+    prevUsage = usage;
+    return percentage;
+}
+
+void Monitor::onAddRecord(uv_timer_t *handle)
+{
+    addRecord();
+}
+
+void Monitor::addRecord()
+{
+    uint64_t percentage = Monitor::getCurrentPercentage();
+    time_t timestamp = time(NULL);
+    timestamps[index] = timestamp;
+    percentages[index] = percentage;
+    index += 1;
+    // std::cout << "addRecord timestamp=" << timestamp << "; percentage=" << percentage << std::endl;
+}
+
+bool Monitor::isTooBusy()
+{
+    time_t now = time(NULL);
+    time_t cutoff = now - WINDOW_SIZE_S;
+    uint64_t total = 0;
+    uint64_t count = 0;
+    uint64_t avg = 0;
+
+    for (int i = 0; i < BUF_SIZE; i++) {
+        time_t timestamp = timestamps[i];
+        uint64_t percentage = percentages[i];
+        if (timestamp >= cutoff) {
+            total += percentage;
+            count += 1;
+        }
+    }
+
+    if (count > 0) {
+        avg = total / count;
+    }
+
+    bool busy = avg > threshold;
+
+    // std::cout << "isTooBusy=" << busy << std::endl;
+    return busy;
+}

src/Monitor.h

@@ -0,0 +1,30 @@
+#ifndef __MONITOR_H__
+#define __MONITOR_H__
+
+#endif
+#include <time.h>
+#include <uv.h>
+
+class Monitor
+{
+public:
+    static const int WINDOW_SIZE_S = 600;
+    static const int BUF_SIZE = 800;
+    static const int INITIAL_WAIT_S = 1;
+    static const int INTERVAL_S = 1;
+
+    static void init(const uint64_t theThreshold);
+    static bool isTooBusy();
+    static uint64_t getCurrentPercentage();
+
+private:
+    static uint64_t threshold;
+    static int index;
+    static uv_timer_t m_timer;
+    static time_t timestamps[BUF_SIZE];
+    static uint64_t percentages[BUF_SIZE];
+    static struct rusage prevUsage;
+
+    static void onAddRecord(uv_timer_t *handle);
+    static void addRecord();        
+};

src/Throttler.cpp

@@ -0,0 +1,26 @@
+#include <iostream>
+#include <uv.h>
+#include "Throttler.h"
+#include "Monitor.h"
+#include "workers/Workers.h"
+
+uint64_t Throttler::interval;
+uv_timer_t Throttler::m_timer;
+
+void Throttler::init(const uint64_t theInterval)
+{
+    interval = theInterval;
+    uv_timer_init(uv_default_loop(), &m_timer);
+    uv_timer_start(&m_timer, Throttler::onCheck, INITIAL_WAIT_S * 1000, interval * 1000);
+}
+
+void Throttler::onCheck(uv_timer_t *handle)
+{
+    if (Workers::isEnabled() && Monitor::isTooBusy()) {
+        // std::cout << "Cooling down workers..." << std::endl;
+        Workers::setEnabled(false);
+    } else if (!Workers::isEnabled() && !Monitor::isTooBusy()) {
+        // std::cout << "Auto-resuming workers..." << std::endl;
+        Workers::setEnabled(true);
+    }
+}

src/Throttler.h

@@ -0,0 +1,19 @@
+#ifndef __THROTTLER_H__
+#define __THROTTLER_H__
+
+#endif
+#include <uv.h>
+
+class Throttler
+{
+public:
+    static const uint64_t INITIAL_WAIT_S = 5;
+
+    static void init(const uint64_t theInterval);
+private:
+    static uint64_t interval;
+    static uv_timer_t m_timer;
+
+    static void onCheck(uv_timer_t *handle);
+};
+