catchunks: switch to RttEstimatorWithStats

Refs: #4887
Change-Id: I8d49eff6ff4a7dc9ff49ffacc0744e50695128d2

tests/chunks/pipeline-interests-aimd.t.cpp

@@ -71,23 +71,23 @@ private:
     return pipelineOptions;
   }
 
-  static RttEstimator::Options
+  static shared_ptr<RttEstimatorWithStats::Options>
   makeRttEstimatorOptions()
   {
-    RttEstimator::Options rttOptions;
-    rttOptions.alpha = 0.125;
-    rttOptions.beta = 0.25;
-    rttOptions.k = 4;
-    rttOptions.initialRto = 1_s;
-    rttOptions.minRto = 200_ms;
-    rttOptions.maxRto = 4_s;
-    rttOptions.rtoBackoffMultiplier = 2;
+    auto rttOptions = make_shared<RttEstimatorWithStats::Options>();
+    rttOptions->alpha = 0.125;
+    rttOptions->beta = 0.25;
+    rttOptions->k = 4;
+    rttOptions->initialRto = 1_s;
+    rttOptions->minRto = 200_ms;
+    rttOptions->maxRto = 4_s;
+    rttOptions->rtoBackoffMultiplier = 2;
     return rttOptions;
   }
 
 protected:
   PipelineInterestsAdaptive::Options opt;
-  RttEstimator rttEstimator;
+  RttEstimatorWithStats rttEstimator;
   PipelineInterestsAdaptive* pipeline;
   static constexpr double MARGIN = 0.001;
 };

tests/chunks/pipeline-interests-cubic.t.cpp

@@ -73,23 +73,23 @@ private:
     return pipelineOptions;
   }
 
-  static RttEstimator::Options
+  static shared_ptr<RttEstimatorWithStats::Options>
   makeRttEstimatorOptions()
   {
-    RttEstimator::Options rttOptions;
-    rttOptions.alpha = 0.125;
-    rttOptions.beta = 0.25;
-    rttOptions.k = 8;
-    rttOptions.initialRto = 1_s;
-    rttOptions.minRto = 200_ms;
-    rttOptions.maxRto = 4_s;
-    rttOptions.rtoBackoffMultiplier = 2;
+    auto rttOptions = make_shared<RttEstimatorWithStats::Options>();
+    rttOptions->alpha = 0.125;
+    rttOptions->beta = 0.25;
+    rttOptions->k = 8;
+    rttOptions->initialRto = 1_s;
+    rttOptions->minRto = 200_ms;
+    rttOptions->maxRto = 4_s;
+    rttOptions->rtoBackoffMultiplier = 2;
     return rttOptions;
   }
 
 protected:
   PipelineInterestsCubic::Options opt;
-  RttEstimator rttEstimator;
+  RttEstimatorWithStats rttEstimator;
   PipelineInterestsCubic* pipeline;
   static constexpr double MARGIN = 0.001;
 };

tools/chunks/catchunks/main.cpp

@@ -195,7 +195,7 @@ main(int argc, char* argv[])
     auto discover = make_unique<DiscoverVersion>(Name(uri), face, options);
     unique_ptr<PipelineInterests> pipeline;
     unique_ptr<StatisticsCollector> statsCollector;
-    unique_ptr<RttEstimator> rttEstimator;
+    unique_ptr<RttEstimatorWithStats> rttEstimator;
     std::ofstream statsFileCwnd;
     std::ofstream statsFileRtt;
 
@@ -205,27 +205,26 @@ main(int argc, char* argv[])
       pipeline = make_unique<PipelineInterestsFixed>(face, optionsPipeline);
     }
     else if (pipelineType == "aimd" || pipelineType == "cubic") {
-      RttEstimator::Options optionsRttEst;
-      optionsRttEst.alpha = rtoAlpha;
-      optionsRttEst.beta = rtoBeta;
-      optionsRttEst.k = k;
-      optionsRttEst.initialRto = 1_s;
-      optionsRttEst.minRto = time::milliseconds(minRto);
-      optionsRttEst.maxRto = time::milliseconds(maxRto);
-      optionsRttEst.rtoBackoffMultiplier = 2;
-      rttEstimator = make_unique<RttEstimator>(optionsRttEst);
-
+      auto optionsRttEst = make_shared<RttEstimatorWithStats::Options>();
+      optionsRttEst->alpha = rtoAlpha;
+      optionsRttEst->beta = rtoBeta;
+      optionsRttEst->k = k;
+      optionsRttEst->initialRto = 1_s;
+      optionsRttEst->minRto = time::milliseconds(minRto);
+      optionsRttEst->maxRto = time::milliseconds(maxRto);
+      optionsRttEst->rtoBackoffMultiplier = 2;
       if (options.isVerbose) {
         using namespace ndn::time;
         std::cerr << "RTT estimator parameters:\n"
-                  << "\tAlpha = " << optionsRttEst.alpha << "\n"
-                  << "\tBeta = " << optionsRttEst.beta << "\n"
-                  << "\tK = " << optionsRttEst.k << "\n"
-                  << "\tInitial RTO = " << duration_cast<milliseconds>(optionsRttEst.initialRto) << "\n"
-                  << "\tMin RTO = " << duration_cast<milliseconds>(optionsRttEst.minRto) << "\n"
-                  << "\tMax RTO = " << duration_cast<milliseconds>(optionsRttEst.maxRto) << "\n"
-                  << "\tBackoff multiplier = " << optionsRttEst.rtoBackoffMultiplier << "\n";
+                  << "\tAlpha = " << optionsRttEst->alpha << "\n"
+                  << "\tBeta = " << optionsRttEst->beta << "\n"
+                  << "\tK = " << optionsRttEst->k << "\n"
+                  << "\tInitial RTO = " << duration_cast<milliseconds>(optionsRttEst->initialRto) << "\n"
+                  << "\tMin RTO = " << duration_cast<milliseconds>(optionsRttEst->minRto) << "\n"
+                  << "\tMax RTO = " << duration_cast<milliseconds>(optionsRttEst->maxRto) << "\n"
+                  << "\tBackoff multiplier = " << optionsRttEst->rtoBackoffMultiplier << "\n";
       }
+      rttEstimator = make_unique<RttEstimatorWithStats>(std::move(optionsRttEst));
 
       PipelineInterestsAdaptive::Options optionsPipeline(options);
       optionsPipeline.disableCwa = disableCwa;
@@ -262,8 +261,7 @@ main(int argc, char* argv[])
             return 4;
           }
         }
-        statsCollector = make_unique<StatisticsCollector>(*adaptivePipeline, *rttEstimator,
-                                                          statsFileCwnd, statsFileRtt);
+        statsCollector = make_unique<StatisticsCollector>(*adaptivePipeline, statsFileCwnd, statsFileRtt);
       }
 
       pipeline = std::move(adaptivePipeline);

tools/chunks/catchunks/pipeline-interests-adaptive.cpp

@@ -37,7 +37,8 @@ namespace chunks {
 
 constexpr double PipelineInterestsAdaptive::MIN_SSTHRESH;
 
-PipelineInterestsAdaptive::PipelineInterestsAdaptive(Face& face, RttEstimator& rttEstimator,
+PipelineInterestsAdaptive::PipelineInterestsAdaptive(Face& face,
+                                                     RttEstimatorWithStats& rttEstimator,
                                                      const Options& options)
   : PipelineInterests(face)
   , m_options(options)
@@ -285,7 +286,11 @@ PipelineInterestsAdaptive::handleData(const Interest& interest, const Data& data
       m_retxCount.count(recvSegNo) == 0) {
     auto nExpectedSamples = std::max<int64_t>((m_nInFlight + 1) >> 1, 1);
     BOOST_ASSERT(nExpectedSamples > 0);
-    m_rttEstimator.addMeasurement(rtt, static_cast<size_t>(nExpectedSamples), recvSegNo);
+    m_rttEstimator.addMeasurement(rtt, static_cast<size_t>(nExpectedSamples));
+    afterRttMeasurement({recvSegNo, rtt,
+                         m_rttEstimator.getSmoothedRtt(),
+                         m_rttEstimator.getRttVariation(),
+                         m_rttEstimator.getEstimatedRto()});
   }
 
   // remove the entry associated with the received segment

tools/chunks/catchunks/pipeline-interests-adaptive.hpp

@@ -40,7 +40,7 @@
 namespace ndn {
 namespace chunks {
 
-using util::RttEstimator;
+using util::RttEstimatorWithStats;
 
 class PipelineInterestsAdaptiveOptions : public Options
 {
@@ -111,7 +111,7 @@ public:
    * Configures the pipelining service without specifying the retrieval namespace. After this
    * configuration the method run must be called to start the Pipeline.
    */
-  PipelineInterestsAdaptive(Face& face, RttEstimator& rttEstimator,
+  PipelineInterestsAdaptive(Face& face, RttEstimatorWithStats& rttEstimator,
                             const Options& options = Options());
 
   ~PipelineInterestsAdaptive() override;
@@ -124,6 +124,20 @@ public:
    */
   signal::Signal<PipelineInterestsAdaptive, time::nanoseconds, double> afterCwndChange;
 
+  struct RttSample
+  {
+    uint64_t segNum;          ///< segment number on which this sample was taken
+    time::nanoseconds rtt;    ///< measured RTT
+    time::nanoseconds sRtt;   ///< smoothed RTT
+    time::nanoseconds rttVar; ///< RTT variation
+    time::nanoseconds rto;    ///< retransmission timeout
+  };
+
+  /**
+   * @brief Signals when a new RTT sample has been taken.
+   */
+  signal::Signal<PipelineInterestsAdaptive, RttSample> afterRttMeasurement;
+
 protected:
   DECLARE_SIGNAL_EMIT(afterCwndChange)
 
@@ -206,7 +220,7 @@ PUBLIC_WITH_TESTS_ELSE_PROTECTED:
   double m_ssthresh; ///< current slow start threshold
 
 PUBLIC_WITH_TESTS_ELSE_PRIVATE:
-  RttEstimator& m_rttEstimator;
+  RttEstimatorWithStats& m_rttEstimator;
   Scheduler m_scheduler;
   scheduler::ScopedEventId m_checkRtoEvent;
 

tools/chunks/catchunks/pipeline-interests-aimd.cpp

@@ -33,7 +33,7 @@
 namespace ndn {
 namespace chunks {
 
-PipelineInterestsAimd::PipelineInterestsAimd(Face& face, RttEstimator& rttEstimator,
+PipelineInterestsAimd::PipelineInterestsAimd(Face& face, RttEstimatorWithStats& rttEstimator,
                                              const Options& options)
   : PipelineInterestsAdaptive(face, rttEstimator, options)
 {

tools/chunks/catchunks/pipeline-interests-aimd.hpp

@@ -40,7 +40,7 @@ namespace chunks {
 class PipelineInterestsAimd : public PipelineInterestsAdaptive
 {
 public:
-  PipelineInterestsAimd(Face& face, RttEstimator& rttEstimator,
+  PipelineInterestsAimd(Face& face, RttEstimatorWithStats& rttEstimator,
                         const Options& options = Options());
 
 private:

tools/chunks/catchunks/pipeline-interests-cubic.cpp

@@ -32,7 +32,7 @@ namespace chunks {
 
 constexpr double CUBIC_C = 0.4;
 
-PipelineInterestsCubic::PipelineInterestsCubic(Face& face, RttEstimator& rttEstimator,
+PipelineInterestsCubic::PipelineInterestsCubic(Face& face, RttEstimatorWithStats& rttEstimator,
                                                const Options& options)
   : PipelineInterestsAdaptive(face, rttEstimator, options)
   , m_cubicOptions(options)

tools/chunks/catchunks/pipeline-interests-cubic.hpp

@@ -61,7 +61,7 @@ public:
   using Options = PipelineInterestsCubicOptions;
 
 public:
-  PipelineInterestsCubic(Face& face, RttEstimator& rttEstimator,
+  PipelineInterestsCubic(Face& face, RttEstimatorWithStats& rttEstimator,
                          const Options& options = Options());
 
 private:

tools/chunks/catchunks/statistics-collector.cpp

@@ -28,25 +28,24 @@ namespace ndn {
 namespace chunks {
 
 StatisticsCollector::StatisticsCollector(PipelineInterestsAdaptive& pipeline,
-                                         RttEstimator& rttEstimator,
                                          std::ostream& osCwnd, std::ostream& osRtt)
   : m_osCwnd(osCwnd)
   , m_osRtt(osRtt)
 {
   m_osCwnd << "time\tcwndsize\n";
   m_osRtt  << "segment\trtt\trttvar\tsrtt\trto\n";
-  pipeline.afterCwndChange.connect(
-    [this] (time::nanoseconds timeElapsed, double cwnd) {
-      m_osCwnd << timeElapsed.count() / 1e9 << '\t' << cwnd << '\n';
-    });
-  rttEstimator.afterMeasurement.connect(
-    [this] (const RttEstimator::Sample& sample) {
-      m_osRtt << *sample.segNum << '\t'
-              << sample.rtt.count() / 1e6 << '\t'
-              << sample.rttVar.count() / 1e6 << '\t'
-              << sample.sRtt.count() / 1e6 << '\t'
-              << sample.rto.count() / 1e6 << '\n';
-    });
+
+  pipeline.afterCwndChange.connect([this] (time::nanoseconds timeElapsed, double cwnd) {
+    m_osCwnd << timeElapsed.count() / 1e9 << '\t' << cwnd << '\n';
+  });
+
+  pipeline.afterRttMeasurement.connect([this] (const auto& sample) {
+    m_osRtt << sample.segNum << '\t'
+            << sample.rtt.count() / 1e6 << '\t'
+            << sample.rttVar.count() / 1e6 << '\t'
+            << sample.sRtt.count() / 1e6 << '\t'
+            << sample.rto.count() / 1e6 << '\n';
+  });
 }
 
 } // namespace chunks

tools/chunks/catchunks/statistics-collector.hpp

@@ -36,7 +36,7 @@ namespace chunks {
 class StatisticsCollector : noncopyable
 {
 public:
-  StatisticsCollector(PipelineInterestsAdaptive& pipeline, RttEstimator& rttEstimator,
+  StatisticsCollector(PipelineInterestsAdaptive& pipeline,
                       std::ostream& osCwnd, std::ostream& osRtt);
 
 private: