Use new RateLimiter in BenchmarkSampler

fairmq/devices/FairMQBenchmarkSampler.cxx

@@ -14,23 +14,22 @@
 
 #include "FairMQBenchmarkSampler.h"
 
-#include <vector>
-#include <chrono>
-
+#include <fairmq/Tools.h>
 #include "../FairMQLogger.h"
 #include "../options/FairMQProgOptions.h"
 
+#include <vector>
+#include <chrono>
+
 using namespace std;
 
 FairMQBenchmarkSampler::FairMQBenchmarkSampler()
     : fSameMessage(true)
     , fMsgSize(10000)
-    , fMsgCounter(0)
-    , fMsgRate(1)
+    , fMsgRate(0)
     , fNumIterations(0)
     , fMaxIterations(0)
     , fOutChannelName()
-    , fResetMsgCounter()
 {
 }
 
@@ -42,16 +41,11 @@ void FairMQBenchmarkSampler::InitTask()
 {
     fSameMessage = fConfig->GetValue<bool>("same-msg");
     fMsgSize = fConfig->GetValue<int>("msg-size");
-    fMsgRate = fConfig->GetValue<int>("msg-rate");
+    fMsgRate = fConfig->GetValue<float>("msg-rate");
     fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
     fOutChannelName = fConfig->GetValue<string>("out-channel");
 }
 
-void FairMQBenchmarkSampler::PreRun()
-{
-    fResetMsgCounter = std::thread(&FairMQBenchmarkSampler::ResetMsgCounter, this);
-}
-
 void FairMQBenchmarkSampler::Run()
 {
     // store the channel reference to avoid traversing the map on every loop iteration
@@ -62,6 +56,8 @@ void FairMQBenchmarkSampler::Run()
     LOG(info) << "Starting the benchmark with message size of " << fMsgSize << " and " << fMaxIterations << " iterations.";
     auto tStart = chrono::high_resolution_clock::now();
 
+    fair::mq::tools::RateLimiter rateLimiter(fMsgRate);
+
     while (CheckCurrentState(RUNNING))
     {
         if (fSameMessage)
@@ -98,31 +94,14 @@ void FairMQBenchmarkSampler::Run()
             }
         }
 
-        --fMsgCounter;
 
-        while (fMsgCounter == 0)
+        if (fMsgRate > 0)
         {
-            this_thread::sleep_for(chrono::milliseconds(1));
+            rateLimiter.maybe_sleep();
         }
     }
 
     auto tEnd = chrono::high_resolution_clock::now();
 
     LOG(info) << "Done " << fNumIterations << " iterations in " << chrono::duration<double, milli>(tEnd - tStart).count() << "ms.";
-
-}
-
-void FairMQBenchmarkSampler::PostRun()
-{
-    fResetMsgCounter.join();
-}
-
-void FairMQBenchmarkSampler::ResetMsgCounter()
-{
-    while (CheckCurrentState(RUNNING))
-    {
-        fMsgCounter = fMsgRate / 100;
-        this_thread::sleep_for(chrono::milliseconds(10));
-    }
-    fMsgCounter = -1;
 }

fairmq/devices/FairMQBenchmarkSampler.h

@@ -31,20 +31,14 @@ class FairMQBenchmarkSampler : public FairMQDevice
     FairMQBenchmarkSampler();
     virtual ~FairMQBenchmarkSampler();
 
-    void PreRun() override;
-    void PostRun() override;
-
-    void ResetMsgCounter();
-
   protected:
     bool fSameMessage;
     int fMsgSize;
     std::atomic<int> fMsgCounter;
-    int fMsgRate;
+    float fMsgRate;
     uint64_t fNumIterations;
     uint64_t fMaxIterations;
     std::string fOutChannelName;
-    std::thread fResetMsgCounter;
 
     virtual void InitTask() override;
     virtual void Run() override;

fairmq/run/runBenchmarkSampler.cxx

@@ -18,7 +18,7 @@ void addCustomOptions(bpo::options_description& options)
         ("same-msg", bpo::value<bool>()->default_value(false), "Re-send the same message, or recreate for each iteration")
         ("msg-size", bpo::value<int>()->default_value(1000000), "Message size in bytes")
         ("max-iterations", bpo::value<uint64_t>()->default_value(0), "Number of run iterations (0 - infinite)")
-        ("msg-rate", bpo::value<int>()->default_value(0), "Msg rate limit in maximum number of messages per second");
+        ("msg-rate", bpo::value<float>()->default_value(0), "Msg rate limit in maximum number of messages per second");
 }
 
 FairMQDevicePtr getDevice(const FairMQProgOptions& /*config*/)