chunks: AIMD congestion control

refs #3636

Change-Id: Ia5e601201219048eb5c745bba9627e4916dac31a

tests/chunks/aimd-rtt-estimator.t.cpp

@@ -0,0 +1,145 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Weiwei Liu
+ */
+
+#include "tools/chunks/catchunks/aimd-rtt-estimator.hpp"
+
+#include "tests/test-common.hpp"
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+namespace tests {
+
+class RttEstimatorFixture
+{
+protected:
+  RttEstimatorFixture()
+    : options(makeRttEstimatorOptions())
+    , rttEstimator(options)
+  {
+  }
+
+private:
+  static RttEstimator::Options
+  makeRttEstimatorOptions()
+  {
+    RttEstimator::Options rttOptions;
+    rttOptions.alpha = 0.125;
+    rttOptions.beta = 0.25;
+    rttOptions.k = 4;
+    rttOptions.minRto = Milliseconds(200.0);
+    rttOptions.maxRto = Milliseconds(4000.0);
+    return rttOptions;
+  }
+
+protected:
+  RttEstimator::Options options;
+  RttEstimator rttEstimator;
+};
+
+BOOST_AUTO_TEST_SUITE(Chunks)
+BOOST_FIXTURE_TEST_SUITE(TestAimdRttEstimator, RttEstimatorFixture)
+
+BOOST_AUTO_TEST_CASE(MeasureRtt)
+{
+  BOOST_REQUIRE(std::isnan(rttEstimator.m_sRtt.count()));
+  BOOST_REQUIRE(std::isnan(rttEstimator.m_rttVar.count()));
+  BOOST_REQUIRE_CLOSE(rttEstimator.m_rto.count(), options.initialRto.count(), 1);
+
+  // first measurement
+  rttEstimator.addMeasurement(1, Milliseconds(100), 1);
+
+  BOOST_CHECK_CLOSE(rttEstimator.m_sRtt.count(), 100, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rttVar.count(), 50, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 300, 1);
+
+  rttEstimator.m_sRtt = Milliseconds(500.0);
+  rttEstimator.m_rttVar = Milliseconds(100.0);
+  rttEstimator.m_rto = Milliseconds(900.0);
+
+  size_t nExpectedSamples = 1;
+  rttEstimator.addMeasurement(1, Milliseconds(100), nExpectedSamples);
+
+  BOOST_CHECK_CLOSE(rttEstimator.m_sRtt.count(), 450, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rttVar.count(), 175, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 1150, 0.1);
+
+  // expected Samples larger than 1
+  nExpectedSamples = 5;
+  rttEstimator.addMeasurement(1, Milliseconds(100), nExpectedSamples);
+
+  BOOST_CHECK_CLOSE(rttEstimator.m_sRtt.count(), 441.25, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rttVar.count(), 183.75, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 1176.25, 0.1);
+
+  rttEstimator.m_sRtt = Milliseconds(100.0);
+  rttEstimator.m_rttVar = Milliseconds(30.0);
+  rttEstimator.m_rto = Milliseconds(220.0);
+
+  // check if minRto works
+  nExpectedSamples = 1;
+  rttEstimator.addMeasurement(1, Milliseconds(100), nExpectedSamples);
+
+  BOOST_CHECK_CLOSE(rttEstimator.m_sRtt.count(), 100, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rttVar.count(), 22.5, 1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 200, 1);
+
+  rttEstimator.m_sRtt = Milliseconds(2000.0);
+  rttEstimator.m_rttVar = Milliseconds(400.0);
+  rttEstimator.m_rto = Milliseconds(3600.0);
+
+  // check if maxRto works
+  nExpectedSamples = 1;
+  rttEstimator.addMeasurement(1, Milliseconds(100), nExpectedSamples);
+
+  BOOST_CHECK_CLOSE(rttEstimator.m_sRtt.count(), 1762.5, 0.1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rttVar.count(), 775, 0.1);
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 4000, 0.1);
+}
+
+BOOST_AUTO_TEST_CASE(RtoBackoff)
+{
+  rttEstimator.m_rto = Milliseconds(500.0);
+  rttEstimator.backoffRto();
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 1000, 0.1);
+
+  // check if minRto works
+  rttEstimator.m_rto = Milliseconds(10.0);
+  rttEstimator.backoffRto();
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 200, 0.1);
+
+  // check if maxRto works
+  rttEstimator.m_rto = Milliseconds(3000.0);
+  rttEstimator.backoffRto();
+  BOOST_CHECK_CLOSE(rttEstimator.m_rto.count(), 4000, 0.1);
+}
+
+BOOST_AUTO_TEST_SUITE_END() // TestAimdRttEstimator
+BOOST_AUTO_TEST_SUITE_END() // Chunks
+
+} // namespace tests
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn

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

@@ -0,0 +1,327 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Weiwei Liu
+ */
+
+#include "tools/chunks/catchunks/pipeline-interests-aimd.hpp"
+#include "tools/chunks/catchunks/options.hpp"
+
+#include "pipeline-interests-fixture.hpp"
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+namespace tests {
+
+using namespace ndn::tests;
+
+class PipelineInterestAimdFixture : public ndn::chunks::tests::PipelineInterestsFixture
+{
+public:
+  PipelineInterestAimdFixture()
+    : PipelineInterestsFixture()
+    , opt(makePipelineOptions())
+    , rttEstimator(makeRttEstimatorOptions())
+  {
+    auto pline = make_unique<PipelineInterestsAimd>(face, rttEstimator, opt);
+    aimdPipeline = pline.get();
+    setPipeline(std::move(pline));
+  }
+
+private:
+  static PipelineInterestsAimdOptions
+  makePipelineOptions()
+  {
+    PipelineInterestsAimdOptions pipelineOptions;
+    pipelineOptions.disableCwa = false;
+    pipelineOptions.resetCwndToInit = false;
+    pipelineOptions.initCwnd = 1.0;
+    pipelineOptions.aiStep = 1.0;
+    pipelineOptions.mdCoef = 0.5;
+    pipelineOptions.initSsthresh = std::numeric_limits<int>::max();
+    return pipelineOptions;
+  }
+
+  static RttEstimator::Options
+  makeRttEstimatorOptions()
+  {
+    RttEstimator::Options rttOptions;
+    rttOptions.alpha = 0.125;
+    rttOptions.beta = 0.25;
+    rttOptions.k = 4;
+    rttOptions.minRto = Milliseconds(200);
+    rttOptions.maxRto = Milliseconds(4000);
+    return rttOptions;
+  }
+
+protected:
+  PipelineInterestsAimdOptions opt;
+  RttEstimator rttEstimator;
+  PipelineInterestsAimd* aimdPipeline;
+};
+
+BOOST_AUTO_TEST_SUITE(Chunks)
+BOOST_FIXTURE_TEST_SUITE(TestPipelineInterestsAimd, PipelineInterestAimdFixture)
+
+BOOST_AUTO_TEST_CASE(SlowStart)
+{
+  nDataSegments = 4;
+  aimdPipeline->m_ssthresh = 8.0;
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 1, 0.1);
+
+  double preCwnd = aimdPipeline->m_cwnd;
+  runWithData(*makeDataWithSegment(0));
+  advanceClocks(io, time::nanoseconds(1));
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 1);
+
+  for (uint64_t i = 1; i < nDataSegments - 1; ++i) {
+    face.receive(*makeDataWithSegment(i));
+    advanceClocks(io, time::nanoseconds(1));
+    BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd - preCwnd, 1, 0.1);
+    preCwnd = aimdPipeline->m_cwnd;
+  }
+}
+
+BOOST_AUTO_TEST_CASE(CongestionAvoidance)
+{
+  nDataSegments = 8;
+  aimdPipeline->m_ssthresh = 4.0;
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 1, 0.1);
+
+  double preCwnd = aimdPipeline->m_cwnd;
+  runWithData(*makeDataWithSegment(0));
+  advanceClocks(io, time::nanoseconds(1));
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 1);
+
+  for (uint64_t i = 1; i < aimdPipeline->m_ssthresh; ++i) { // slow start
+    face.receive(*makeDataWithSegment(i));
+    advanceClocks(io, time::nanoseconds(1));
+    preCwnd = aimdPipeline->m_cwnd;
+  }
+
+  BOOST_REQUIRE_CLOSE(preCwnd, aimdPipeline->m_ssthresh, 0.1);
+
+  for (uint64_t i = aimdPipeline->m_ssthresh; i < nDataSegments - 1; ++i) { // congestion avoidance
+    face.receive(*makeDataWithSegment(i));
+    advanceClocks(io, time::nanoseconds(1));
+    BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd - preCwnd, opt.aiStep / floor(aimdPipeline->m_cwnd), 0.1);
+    preCwnd = aimdPipeline->m_cwnd;
+  }
+}
+
+BOOST_AUTO_TEST_CASE(Timeout)
+{
+  nDataSegments = 8;
+  aimdPipeline->m_ssthresh = 4.0;
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 1, 0.1);
+
+  runWithData(*makeDataWithSegment(0));
+  advanceClocks(io, time::nanoseconds(1));
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 1);
+
+  // receive segment 1 and segment 2
+  for (uint64_t i = 1; i < 3; ++i) {
+    face.receive(*makeDataWithSegment(i));
+    advanceClocks(io, time::nanoseconds(1));
+  }
+
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 3, 0.1);
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 5); // request for segment 5 has been sent
+
+  advanceClocks(io, time::milliseconds(100));
+
+  // receive segment 4
+  face.receive(*makeDataWithSegment(4));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // receive segment 5
+  face.receive(*makeDataWithSegment(5));
+  advanceClocks(io, time::nanoseconds(1));
+
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 4.25, 0.1);
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 7); // all the segment requests have been sent
+
+  // timeout segment 3
+  advanceClocks(io, time::milliseconds(150));
+
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 2.125, 0.1); // window size drop to 1/2 of previous size
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxQueue.size(), 1);
+
+  // receive segment 6, retransmit 3
+  face.receive(*makeDataWithSegment(6));
+  advanceClocks(io, time::nanoseconds(1));
+
+  BOOST_REQUIRE_CLOSE(aimdPipeline->m_cwnd, 2.625, 0.1); // congestion avoidance
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxQueue.size(), 0);
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxCount[3], 1);
+}
+
+BOOST_AUTO_TEST_CASE(Nack)
+{
+  nDataSegments = 5;
+  aimdPipeline->m_cwnd = 10.0;
+  runWithData(*makeDataWithSegment(0));
+  advanceClocks(io, time::nanoseconds(1));
+
+  face.receive(*makeDataWithSegment(1));
+  advanceClocks(io, time::nanoseconds(1));
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 4);
+
+  // receive a nack with NackReason::DUPLICATE for segment 2
+  auto nack1 = makeNack(face.sentInterests[1], lp::NackReason::DUPLICATE);
+  face.receive(nack1);
+  advanceClocks(io, time::nanoseconds(1));
+
+  // nack1 is ignored
+  BOOST_CHECK_EQUAL(hasFailed, false);
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxQueue.size(), 0);
+
+  // receive a nack with NackReason::CONGESTION for segment 3
+  auto nack2 = makeNack(face.sentInterests[2], lp::NackReason::CONGESTION);
+  face.receive(nack2);
+  advanceClocks(io, time::nanoseconds(1));
+
+  // segment 3 is retransmitted
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxCount[3], 1);
+
+  // receive a nack with NackReason::NONE for segment 4
+  auto nack3 = makeNack(face.sentInterests[3], lp::NackReason::NONE);
+  face.receive(nack3);
+  advanceClocks(io, time::nanoseconds(1));
+
+  // Other types of Nack will trigger a failure
+  BOOST_CHECK_EQUAL(hasFailed, true);
+}
+
+BOOST_AUTO_TEST_CASE(FinalBlockIdNotSetAtBeginning)
+{
+  nDataSegments = 4;
+  aimdPipeline->m_cwnd = 4;
+  runWithData(*makeDataWithSegment(0, false));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // receive segment 1 without FinalBlockId
+  face.receive(*makeDataWithSegment(1, false));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // interests for segment 1 - 6 have been sent
+  BOOST_CHECK_EQUAL(face.sentInterests.size(), 6);
+  BOOST_CHECK_EQUAL(aimdPipeline->m_hasFinalBlockId, false);
+  // pending interests: segment 2, 3, 4, 5, 6
+  BOOST_CHECK_EQUAL(face.getNPendingInterests(), 5);
+
+  // receive segment 2 with FinalBlockId
+  face.receive(*makeDataWithSegment(2));
+  advanceClocks(io, time::nanoseconds(1));
+  BOOST_CHECK_EQUAL(aimdPipeline->m_hasFinalBlockId, true);
+
+  // pending interests for segment 2, 4, 5, 6 haven been removed
+  BOOST_CHECK_EQUAL(face.getNPendingInterests(), 1);
+}
+
+BOOST_AUTO_TEST_CASE(FailureBeforeFinalBlockIdReceived)
+{
+  // failed to retrieve segNo while the FinalBlockId has not yet been
+  // set, and later received a FinalBlockId >= segNo, i.e. segNo is
+  // part of the content.
+
+  nDataSegments = 4;
+  aimdPipeline->m_cwnd = 4;
+  runWithData(*makeDataWithSegment(0, false));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // receive segment 1 without FinalBlockId
+  face.receive(*makeDataWithSegment(1, false));
+  advanceClocks(io, time::nanoseconds(1));
+  // interests for segment 1 - 6 have been sent
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 6);
+
+  // receive nack with NackReason::NONE for segment 3
+  auto nack = makeNack(face.sentInterests[2], lp::NackReason::NONE);
+  face.receive(nack);
+  advanceClocks(io, time::nanoseconds(1));
+
+  // error not triggered
+  // pending interests for segment > 3 haven been removed
+  BOOST_CHECK_EQUAL(hasFailed, false);
+  BOOST_CHECK_EQUAL(face.getNPendingInterests(), 1);
+
+  // receive segment 2 with FinalBlockId
+  face.receive(*makeDataWithSegment(2));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // error triggered since segment 3 is part of the content
+  BOOST_CHECK_EQUAL(hasFailed, true);
+}
+
+BOOST_AUTO_TEST_CASE(SpuriousFailureBeforeFinalBlockIdReceived)
+{
+  // failed to retrieve segNo while the FinalBlockId has not yet been
+  // set, and later received a FinalBlockId < segNo, i.e. segNo is
+  // not part of the content, and it was actually a spurious failure
+
+  nDataSegments = 4;
+  aimdPipeline->m_cwnd = 4;
+  runWithData(*makeDataWithSegment(0, false));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // receive segment 1 without FinalBlockId
+  face.receive(*makeDataWithSegment(1, false));
+  advanceClocks(io, time::nanoseconds(1));
+  // interests for segment 1 - 6 have been sent
+  BOOST_REQUIRE_EQUAL(face.sentInterests.size(), 6);
+
+  // receive nack with NackReason::NONE for segment 4
+  auto nack= makeNack(face.sentInterests[3], lp::NackReason::NONE);
+  face.receive(nack);
+  advanceClocks(io, time::nanoseconds(1));
+
+  // error not triggered
+  // pending interests for segment > 4 haven been removed
+  BOOST_CHECK_EQUAL(hasFailed, false);
+  BOOST_CHECK_EQUAL(face.getNPendingInterests(), 2);
+
+  // receive segment 2 with FinalBlockId
+  face.receive(*makeDataWithSegment(2));
+  advanceClocks(io, time::nanoseconds(1));
+
+  // timeout segment 3
+  advanceClocks(io, time::milliseconds(250));
+
+  // segment 3 is retransmitted
+  BOOST_CHECK_EQUAL(aimdPipeline->m_retxCount[3], 1);
+
+  // receive segment 3
+  face.receive(*makeDataWithSegment(3));
+  advanceClocks(io, time::nanoseconds(1));
+
+  BOOST_CHECK_EQUAL(hasFailed, false);
+}
+
+BOOST_AUTO_TEST_SUITE_END() // TestPipelineInterestsAimd
+BOOST_AUTO_TEST_SUITE_END() // Chunks
+
+} // namespace tests
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn

tools/chunks/README.md

@@ -11,13 +11,13 @@
   discovering the latest version of the file, and writes the content of the retrieved file to
   the standard output.
 
-## Version discovery methods
+## Version discovery methods in ndncatchunks
 
-* `fixed`    : ndncatchunks will send an interest attempting to find a data packet with the
+* `fixed`    : sends an interest attempting to find a data packet with the
                specified prefix and version number. A version component must be present at the
                end of the user-specified NDN name.
 
-* `iterative`: ndncatchunks will send a series of interests with ChildSelector set to prefer the
+* `iterative`: sends a series of interests with ChildSelector set to prefer the
                rightmost child and Exclude selectors, attempting to find a data packet with the
                specified prefix and the latest (the largest in the NDN canonical ordering)
                version number.  The version is declared "latest" after a predefined number of
@@ -25,6 +25,19 @@
 
 The default discovery method is `iterative`.
 
+## Interest pipeline types in ndncatchunks
+
+* `fixed`: maintains a fixed-size window of Interests in flight; the window size is configurable
+           via a command line option and defaults to 1.
+
+* `aimd` : sends Interests using an additive-increase/multiplicative-decrease (AIMD) algorithm to
+           control the window size. By default, a Conservative Loss Adaptation algorithm is adopted
+           combining with the AIMD algorithm, that is, at most one window decrease will be
+           performed per round-trip-time. For details please refer to:
+  [A Practical Congestion Control Scheme for Named Data
+  Networking](https://www.researchgate.net/publication/306259672_A_Practical_Congestion_Control_Scheme_for_Named_Data_Networking)
+
+The default Interest pipeline type is `fixed`.
 
 ## Usage examples
 

tools/chunks/catchunks/aimd-rtt-estimator.cpp

@@ -0,0 +1,87 @@
+/**
+ * Copyright (c) 2016,  Arizona Board of Regents.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Shuo Yang
+ * @author Weiwei Liu
+ */
+
+#include "aimd-rtt-estimator.hpp"
+#include <cmath>
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+RttEstimator::RttEstimator(const Options& options)
+  : m_options(options)
+  , m_sRtt(std::numeric_limits<double>::quiet_NaN())
+  , m_rttVar(std::numeric_limits<double>::quiet_NaN())
+  , m_rto(m_options.initialRto.count())
+{
+  if (m_options.isVerbose) {
+    std::cerr << m_options;
+  }
+}
+
+void
+RttEstimator::addMeasurement(uint64_t segNo, Milliseconds rtt, size_t nExpectedSamples)
+{
+  BOOST_ASSERT(nExpectedSamples > 0);
+
+  if (std::isnan(m_sRtt.count())) { // first measurement
+    m_sRtt = rtt;
+    m_rttVar = m_sRtt / 2;
+    m_rto = m_sRtt + m_options.k * m_rttVar;
+  }
+  else {
+    double alpha = m_options.alpha / nExpectedSamples;
+    double beta = m_options.beta / nExpectedSamples;
+    m_rttVar = (1 - beta) * m_rttVar + beta * time::abs(m_sRtt - rtt);
+    m_sRtt = (1 - alpha) * m_sRtt + alpha * rtt;
+    m_rto = m_sRtt + m_options.k * m_rttVar;
+  }
+
+  m_rto = ndn::clamp(m_rto, m_options.minRto, m_options.maxRto);
+
+  afterRttMeasurement({segNo, rtt, m_sRtt, m_rttVar, m_rto});
+}
+
+void
+RttEstimator::backoffRto()
+{
+  m_rto = ndn::clamp(m_rto * m_options.rtoBackoffMultiplier,
+                     m_options.minRto, m_options.maxRto);
+}
+
+std::ostream&
+operator<<(std::ostream& os, const RttEstimator::Options& options)
+{
+  os << "RttEstimator initial parameters:\n"
+     << "\tAlpha = " << options.alpha << "\n"
+     << "\tBeta = " << options.beta << "\n"
+     << "\tK = " << options.k << "\n"
+     << "\tInitial RTO = " << options.initialRto << "\n"
+     << "\tMin RTO = " << options.minRto << "\n"
+     << "\tMax RTO = " << options.maxRto << "\n";
+  return os;
+}
+
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn

tools/chunks/catchunks/aimd-rtt-estimator.hpp

@@ -0,0 +1,141 @@
+/**
+ * Copyright (c) 2016,  Arizona Board of Regents.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Shuo Yang
+ * @author Weiwei Liu
+ */
+
+#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_AIMD_RTT_ESTIMATOR_HPP
+#define NDN_TOOLS_CHUNKS_CATCHUNKS_AIMD_RTT_ESTIMATOR_HPP
+
+#include "core/common.hpp"
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+typedef time::duration<double, time::milliseconds::period> Milliseconds;
+
+struct RttRtoSample
+{
+  uint64_t segNo;
+  Milliseconds rtt; ///< measured RTT
+  Milliseconds sRtt; ///< smoothed RTT
+  Milliseconds rttVar; ///< RTT variation
+  Milliseconds rto; ///< retransmission timeout
+};
+
+/**
+ * @brief RTT Estimator.
+ *
+ * This class implements the "Mean--Deviation" RTT estimator, as discussed in RFC6298,
+ * with the modifications to RTO calculation described in RFC 7323 Appendix G.
+ */
+class RttEstimator
+{
+public:
+  class Options
+  {
+  public:
+    Options()
+      : isVerbose(false)
+      , alpha(0.125)
+      , beta(0.25)
+      , k(4)
+      , initialRto(1000.0)
+      , minRto(200.0)
+      , maxRto(20000.0)
+      , rtoBackoffMultiplier(2)
+    {
+    }
+
+  public:
+    bool isVerbose;
+    double alpha; ///< parameter for RTT estimation
+    double beta; ///< parameter for RTT variation calculation
+    int k; ///< factor of RTT variation when calculating RTO
+    Milliseconds initialRto; ///< initial RTO value
+    Milliseconds minRto; ///< lower bound of RTO
+    Milliseconds maxRto; ///< upper bound of RTO
+    int rtoBackoffMultiplier;
+  };
+
+  /**
+   * @brief create a RTT Estimator
+   *
+   * Configures the RTT Estimator with the default parameters if an instance of Options
+   * is not passed to the constructor.
+   */
+  explicit
+  RttEstimator(const Options& options = Options());
+
+  /**
+   * @brief Add a new RTT measurement to the estimator for the given received segment.
+   *
+   * @note Don't take RTT measurement for retransmitted segments
+   * @param segNo the segment number of the received segmented Data
+   * @param rtt the sampled rtt
+   * @param nExpectedSamples number of expected samples, must be greater than 0.
+   *        It should be set to current number of in-flight Interests. Please
+   *        refer to Appendix G of RFC 7323 for details.
+   */
+  void
+  addMeasurement(uint64_t segNo, Milliseconds rtt, size_t nExpectedSamples);
+
+  /**
+   * @return estimated RTO
+   */
+  Milliseconds
+  getEstimatedRto() const;
+
+  /**
+   * @brief backoff RTO by the factor of RttEstimatorOptions::rtoBackoffMultiplier
+   */
+  void
+  backoffRto();
+
+  /**
+   * @brief Signals after rtt is measured
+   */
+  signal::Signal<RttEstimator, RttRtoSample> afterRttMeasurement;
+
+PUBLIC_WITH_TESTS_ELSE_PRIVATE:
+  const Options m_options;
+  Milliseconds m_sRtt; ///< smoothed round-trip time
+  Milliseconds m_rttVar; ///< round-trip time variation
+  Milliseconds m_rto; ///< retransmission timeout
+};
+
+/**
+ * @brief returns the estimated RTO value
+ */
+inline Milliseconds
+RttEstimator::getEstimatedRto() const
+{
+  return m_rto;
+}
+
+std::ostream&
+operator<<(std::ostream& os, const RttEstimator::Options& options);
+
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn
+
+#endif // NDN_TOOLS_CHUNKS_CATCHUNKS_RTT_ESTIMATOR_HPP

tools/chunks/catchunks/aimd-statistics-collector.cpp

@@ -0,0 +1,54 @@
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Weiwei Liu
+ */
+
+#include "aimd-statistics-collector.hpp"
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+StatisticsCollector::StatisticsCollector(PipelineInterestsAimd& 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] (Milliseconds timeElapsed, double cwnd) {
+      m_osCwnd << timeElapsed.count() / 1000 << '\t' << cwnd << '\n';
+    });
+  rttEstimator.afterRttMeasurement.connect(
+    [this] (const RttRtoSample& rttSample) {
+      m_osRtt << rttSample.segNo << '\t'
+              << rttSample.rtt.count() << '\t'
+              << rttSample.rttVar.count() << '\t'
+              << rttSample.sRtt.count() << '\t'
+              << rttSample.rto.count() << '\n';
+    });
+}
+
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn

tools/chunks/catchunks/aimd-statistics-collector.hpp

@@ -0,0 +1,53 @@
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Weiwei Liu
+ */
+
+#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_AIMD_STATISTICS_COLLECTOR_HPP
+#define NDN_TOOLS_CHUNKS_CATCHUNKS_AIMD_STATISTICS_COLLECTOR_HPP
+
+#include "pipeline-interests-aimd.hpp"
+#include "aimd-rtt-estimator.hpp"
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+/**
+ * @brief Statistics collector for AIMD pipeline
+ */
+class StatisticsCollector : noncopyable
+{
+public:
+  StatisticsCollector(PipelineInterestsAimd& pipeline, RttEstimator& rttEstimator,
+                      std::ostream& osCwnd, std::ostream& osRtt);
+
+private:
+  std::ostream& m_osCwnd;
+  std::ostream& m_osRtt;
+};
+
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn
+
+#endif // NDN_TOOLS_CHUNKS_CATCHUNKS_AIMD_STATISTICS_COLLECTOR_HPP

tools/chunks/catchunks/ndncatchunks.cpp

@@ -33,8 +33,12 @@
 #include "discover-version-fixed.hpp"
 #include "discover-version-iterative.hpp"
 #include "pipeline-interests-fixed-window.hpp"
+#include "pipeline-interests-aimd.hpp"
+#include "aimd-rtt-estimator.hpp"
+#include "aimd-statistics-collector.hpp"
 
 #include <ndn-cxx/security/validator-null.hpp>
+#include <fstream>
 
 namespace ndn {
 namespace chunks {
@@ -50,34 +54,87 @@ main(int argc, char** argv)
   int maxRetriesAfterVersionFound(1);
   std::string uri;
 
+  // congestion control parameters, CWA refers to conservative window adaptation,
+  // i.e. only reduce window size at most once per RTT
+  bool disableCwa(false), resetCwndToInit(false);
+  double aiStep(1.0), mdCoef(0.5), alpha(0.125), beta(0.25),
+         minRto(200.0), maxRto(4000.0);
+  int initCwnd(1), initSsthresh(std::numeric_limits<int>::max()), k(4);
+  std::string cwndPath, rttPath;
+
   namespace po = boost::program_options;
-  po::options_description visibleDesc("Options");
-  visibleDesc.add_options()
+  po::options_description basicDesc("Basic Options");
+  basicDesc.add_options()
     ("help,h",      "print this help message and exit")
     ("discover-version,d",  po::value<std::string>(&discoverType)->default_value(discoverType),
                             "version discovery algorithm to use; valid values are: 'fixed', 'iterative'")
+    ("pipeline-type,t",  po::value<std::string>(&pipelineType)->default_value(pipelineType),
+                         "type of Interest pipeline to use; valid values are: 'fixed', 'aimd'")
     ("fresh,f",     po::bool_switch(&options.mustBeFresh), "only return fresh content")
     ("lifetime,l",  po::value<uint64_t>()->default_value(options.interestLifetime.count()),
                     "lifetime of expressed Interests, in milliseconds")
-    ("pipeline,p",  po::value<size_t>(&maxPipelineSize)->default_value(maxPipelineSize),
-                    "maximum size of the Interest pipeline")
     ("retries,r",   po::value<int>(&options.maxRetriesOnTimeoutOrNack)->default_value(options.maxRetriesOnTimeoutOrNack),
                     "maximum number of retries in case of Nack or timeout (-1 = no limit)")
-    ("retries-iterative,i", po::value<int>(&maxRetriesAfterVersionFound)->default_value(maxRetriesAfterVersionFound),
-                            "number of timeouts that have to occur in order to confirm a discovered Data "
-                            "version as the latest one (only applicable to 'iterative' version discovery)")
     ("verbose,v",   po::bool_switch(&options.isVerbose), "turn on verbose output")
     ("version,V",   "print program version and exit")
     ;
 
-  po::options_description hiddenDesc("Hidden options");
+  po::options_description iterDiscoveryDesc("Iterative version discovery options");
+  iterDiscoveryDesc.add_options()
+    ("retries-iterative,i", po::value<int>(&maxRetriesAfterVersionFound)->default_value(maxRetriesAfterVersionFound),
+                            "number of timeouts that have to occur in order to confirm a discovered Data "
+                            "version as the latest one")
+    ;
+
+  po::options_description fixedPipeDesc("Fixed pipeline options");
+  fixedPipeDesc.add_options()
+    ("pipeline-size,s", po::value<size_t>(&maxPipelineSize)->default_value(maxPipelineSize),
+                        "size of the Interest pipeline")
+    ;
+
+  po::options_description aimdPipeDesc("AIMD pipeline options");
+  aimdPipeDesc.add_options()
+    ("aimd-debug-cwnd", po::value<std::string>(&cwndPath),
+                        "log file for AIMD cwnd statistics")
+    ("aimd-debug-rtt", po::value<std::string>(&rttPath),
+                       "log file for AIMD rtt statistics")
+    ("aimd-disable-cwa", po::bool_switch(&disableCwa),
+                         "disable Conservative Window Adaptation, "
+                         "i.e. reduce window on each timeout (instead of at most once per RTT)")
+    ("aimd-reset-cwnd-to-init", po::bool_switch(&resetCwndToInit),
+                                "reset cwnd to initial cwnd when loss event occurs, default is "
+                                "resetting to ssthresh")
+    ("aimd-initial-cwnd",       po::value<int>(&initCwnd)->default_value(initCwnd),
+                                "initial cwnd")
+    ("aimd-initial-ssthresh",   po::value<int>(&initSsthresh),
+                                "initial slow start threshold (defaults to infinity)")
+    ("aimd-aistep",    po::value<double>(&aiStep)->default_value(aiStep),
+                       "additive-increase step")
+    ("aimd-mdcoef",    po::value<double>(&mdCoef)->default_value(mdCoef),
+                       "multiplicative-decrease coefficient")
+    ("aimd-rto-alpha", po::value<double>(&alpha)->default_value(alpha),
+                       "alpha value for rto calculation")
+    ("aimd-rto-beta",  po::value<double>(&beta)->default_value(beta),
+                       "beta value for rto calculation")
+    ("aimd-rto-k",     po::value<int>(&k)->default_value(k),
+                       "k value for rto calculation")
+    ("aimd-rto-min",   po::value<double>(&minRto)->default_value(minRto),
+                       "min rto value in milliseconds")
+    ("aimd-rto-max",   po::value<double>(&maxRto)->default_value(maxRto),
+                       "max rto value in milliseconds")
+    ;
+
+  po::options_description visibleDesc;
+  visibleDesc.add(basicDesc).add(iterDiscoveryDesc).add(fixedPipeDesc).add(aimdPipeDesc);
+
+  po::options_description hiddenDesc;
   hiddenDesc.add_options()
     ("ndn-name,n", po::value<std::string>(&uri), "NDN name of the requested content");
 
   po::positional_options_description p;
   p.add("ndn-name", -1);
 
-  po::options_description optDesc("Allowed options");
+  po::options_description optDesc;
   optDesc.add(visibleDesc).add(hiddenDesc);
 
   po::variables_map vm;
@@ -153,11 +210,59 @@ main(int argc, char** argv)
     }
 
     unique_ptr<PipelineInterests> pipeline;
+    unique_ptr<aimd::StatisticsCollector> statsCollector;
+    unique_ptr<aimd::RttEstimator> rttEstimator;
+    std::ofstream statsFileCwnd;
+    std::ofstream statsFileRtt;
+
     if (pipelineType == "fixed") {
       PipelineInterestsFixedWindow::Options optionsPipeline(options);
       optionsPipeline.maxPipelineSize = maxPipelineSize;
       pipeline = make_unique<PipelineInterestsFixedWindow>(face, optionsPipeline);
     }
+    else if (pipelineType == "aimd") {
+      aimd::RttEstimator::Options optionsRttEst;
+      optionsRttEst.isVerbose = options.isVerbose;
+      optionsRttEst.alpha = alpha;
+      optionsRttEst.beta = beta;
+      optionsRttEst.k = k;
+      optionsRttEst.minRto = aimd::Milliseconds(minRto);
+      optionsRttEst.maxRto = aimd::Milliseconds(maxRto);
+
+      rttEstimator = make_unique<aimd::RttEstimator>(optionsRttEst);
+
+      PipelineInterestsAimd::Options optionsPipeline;
+      optionsPipeline.isVerbose = options.isVerbose;
+      optionsPipeline.disableCwa = disableCwa;
+      optionsPipeline.resetCwndToInit = resetCwndToInit;
+      optionsPipeline.initCwnd = static_cast<double>(initCwnd);
+      optionsPipeline.initSsthresh = static_cast<double>(initSsthresh);
+      optionsPipeline.aiStep = aiStep;
+      optionsPipeline.mdCoef = mdCoef;
+
+      auto aimdPipeline = make_unique<PipelineInterestsAimd>(face, *rttEstimator, optionsPipeline);
+
+      if (!cwndPath.empty() || !rttPath.empty()) {
+        if (!cwndPath.empty()) {
+          statsFileCwnd.open(cwndPath);
+          if (statsFileCwnd.fail()) {
+            std::cerr << "ERROR: failed to open " << cwndPath << std::endl;
+            return 4;
+          }
+        }
+        if (!rttPath.empty()) {
+          statsFileRtt.open(rttPath);
+          if (statsFileRtt.fail()) {
+            std::cerr << "ERROR: failed to open " << rttPath << std::endl;
+            return 4;
+          }
+        }
+        statsCollector = make_unique<aimd::StatisticsCollector>(*aimdPipeline, *rttEstimator,
+                                                                statsFileCwnd, statsFileRtt);
+      }
+
+      pipeline = std::move(aimdPipeline);
+    }
     else {
       std::cerr << "ERROR: Interest pipeline type not valid" << std::endl;
       return 2;

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

@@ -0,0 +1,509 @@
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Shuo Yang
+ * @author Weiwei Liu
+ */
+
+#include "pipeline-interests-aimd.hpp"
+
+#include <cmath>
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+PipelineInterestsAimd::PipelineInterestsAimd(Face& face, RttEstimator& rttEstimator,
+                                             const Options& options)
+  : PipelineInterests(face)
+  , m_options(options)
+  , m_rttEstimator(rttEstimator)
+  , m_scheduler(m_face.getIoService())
+  , m_nextSegmentNo(0)
+  , m_receivedSize(0)
+  , m_highData(0)
+  , m_highInterest(0)
+  , m_recPoint(0)
+  , m_nInFlight(0)
+  , m_nReceived(0)
+  , m_nLossEvents(0)
+  , m_nRetransmitted(0)
+  , m_cwnd(m_options.initCwnd)
+  , m_ssthresh(m_options.initSsthresh)
+  , m_hasFailure(false)
+  , m_failedSegNo(0)
+{
+  if (m_options.isVerbose) {
+    std::cerr << m_options;
+  }
+}
+
+PipelineInterestsAimd::~PipelineInterestsAimd()
+{
+  cancel();
+}
+
+void
+PipelineInterestsAimd::doRun()
+{
+  // record the start time of running pipeline
+  m_startTime = time::steady_clock::now();
+
+  // count the excluded segment
+  m_nReceived++;
+
+  // schedule the event to check retransmission timer
+  m_scheduler.scheduleEvent(m_options.rtoCheckInterval, [this] { checkRto(); });
+
+  sendInterest(getNextSegmentNo(), false);
+}
+
+void
+PipelineInterestsAimd::doCancel()
+{
+  for (const auto& entry : m_segmentInfo) {
+    const SegmentInfo& segInfo = entry.second;
+    m_face.removePendingInterest(segInfo.interestId);
+  }
+  m_segmentInfo.clear();
+  m_scheduler.cancelAllEvents();
+}
+
+void
+PipelineInterestsAimd::checkRto()
+{
+  if (isStopping())
+    return;
+
+  int timeoutCount = 0;
+
+  for (auto& entry : m_segmentInfo) {
+    SegmentInfo& segInfo = entry.second;
+    if (segInfo.state != SegmentState::InRetxQueue && // do not check segments currently in the retx queue
+        segInfo.state != SegmentState::RetxReceived) { // or already-received retransmitted segments
+      Milliseconds timeElapsed = time::steady_clock::now() - segInfo.timeSent;
+      if (timeElapsed.count() > segInfo.rto.count()) { // timer expired?
+        uint64_t timedoutSeg = entry.first;
+        m_retxQueue.push(timedoutSeg); // put on retx queue
+        segInfo.state = SegmentState::InRetxQueue; // update status
+        timeoutCount++;
+      }
+    }
+  }
+
+  if (timeoutCount > 0) {
+    handleTimeout(timeoutCount);
+  }
+
+  // schedule the next check after predefined interval
+  m_scheduler.scheduleEvent(m_options.rtoCheckInterval, [this] { checkRto(); });
+}
+
+void
+PipelineInterestsAimd::sendInterest(uint64_t segNo, bool isRetransmission)
+{
+  if (isStopping())
+    return;
+
+  if (m_hasFinalBlockId && segNo > m_lastSegmentNo && !isRetransmission)
+    return;
+
+  if (!isRetransmission && m_hasFailure)
+    return;
+
+  if (m_options.isVerbose) {
+    if (isRetransmission)
+      std::cerr << "Retransmitting segment #" << segNo << std::endl;
+    else
+      std::cerr << "Requesting segment #" << segNo << std::endl;
+  }
+
+  if (isRetransmission) {
+    auto ret = m_retxCount.insert(std::make_pair(segNo, 1));
+    if (ret.second == false) { // not the first retransmission
+      m_retxCount[segNo] += 1;
+      if (m_retxCount[segNo] > m_options.maxRetriesOnTimeoutOrNack) {
+        return handleFail(segNo, "Reached the maximum number of retries (" +
+                          to_string(m_options.maxRetriesOnTimeoutOrNack) +
+                          ") while retrieving segment #" + to_string(segNo));
+      }
+
+      if (m_options.isVerbose) {
+        std::cerr << "# of retries for segment #" << segNo
+                  << " is " << m_retxCount[segNo] << std::endl;
+      }
+    }
+
+    m_face.removePendingInterest(m_segmentInfo[segNo].interestId);
+  }
+
+  Interest interest(Name(m_prefix).appendSegment(segNo));
+  interest.setInterestLifetime(m_options.interestLifetime);
+  interest.setMustBeFresh(m_options.mustBeFresh);
+  interest.setMaxSuffixComponents(1);
+
+  auto interestId = m_face.expressInterest(interest,
+                                           bind(&PipelineInterestsAimd::handleData, this, _1, _2),
+                                           bind(&PipelineInterestsAimd::handleNack, this, _1, _2),
+                                           bind(&PipelineInterestsAimd::handleLifetimeExpiration,
+                                                this, _1));
+
+  m_nInFlight++;
+
+  if (isRetransmission) {
+    SegmentInfo& segInfo = m_segmentInfo[segNo];
+    segInfo.state = SegmentState::Retransmitted;
+    segInfo.rto = m_rttEstimator.getEstimatedRto();
+    segInfo.timeSent = time::steady_clock::now();
+    m_nRetransmitted++;
+  }
+  else {
+    m_highInterest = segNo;
+    Milliseconds rto = m_rttEstimator.getEstimatedRto();
+    SegmentInfo segInfo{interestId, SegmentState::FirstTimeSent, rto, time::steady_clock::now()};
+
+    m_segmentInfo.emplace(segNo, segInfo);
+  }
+}
+
+void
+PipelineInterestsAimd::schedulePackets()
+{
+  int availableWindowSize = static_cast<int>(m_cwnd) - m_nInFlight;
+  while (availableWindowSize > 0) {
+    if (!m_retxQueue.empty()) { // do retransmission first
+      uint64_t retxSegNo = m_retxQueue.front();
+      m_retxQueue.pop();
+
+      auto it = m_segmentInfo.find(retxSegNo);
+      if (it == m_segmentInfo.end()) {
+        continue;
+      }
+      // the segment is still in the map, it means that it needs to be retransmitted
+      sendInterest(retxSegNo, true);
+    }
+    else { // send next segment
+      sendInterest(getNextSegmentNo(), false);
+    }
+    availableWindowSize--;
+  }
+}
+
+void
+PipelineInterestsAimd::handleData(const Interest& interest, const Data& data)
+{
+  if (isStopping())
+    return;
+
+  // Data name will not have extra components because MaxSuffixComponents is set to 1
+  BOOST_ASSERT(data.getName().equals(interest.getName()));
+
+  if (!m_hasFinalBlockId && !data.getFinalBlockId().empty()) {
+    m_lastSegmentNo = data.getFinalBlockId().toSegment();
+    m_hasFinalBlockId = true;
+    cancelInFlightSegmentsGreaterThan(m_lastSegmentNo);
+    if (m_hasFailure && m_lastSegmentNo >= m_failedSegNo) {
+      // previously failed segment is part of the content
+      return onFailure(m_failureReason);
+    } else {
+      m_hasFailure = false;
+    }
+  }
+
+  uint64_t recvSegNo = data.getName()[-1].toSegment();
+  if (m_highData < recvSegNo) {
+    m_highData = recvSegNo;
+  }
+
+  SegmentInfo& segInfo = m_segmentInfo[recvSegNo];
+  if (segInfo.state == SegmentState::RetxReceived) {
+    m_segmentInfo.erase(recvSegNo);
+    return; // ignore already-received segment
+  }
+
+  Milliseconds rtt = time::steady_clock::now() - segInfo.timeSent;
+
+  if (m_options.isVerbose) {
+    std::cerr << "Received segment #" << recvSegNo
+              << ", rtt=" << rtt.count() << "ms"
+              << ", rto=" << segInfo.rto.count() << "ms" << std::endl;
+  }
+
+  // for segments in retransmission queue, no need to decrement m_nInFlight since
+  // it's already been decremented when segments timed out
+  if (segInfo.state != SegmentState::InRetxQueue && m_nInFlight > 0) {
+    m_nInFlight--;
+  }
+
+  m_receivedSize += data.getContent().value_size();
+  m_nReceived++;
+
+  increaseWindow();
+  onData(interest, data);
+
+  if (segInfo.state == SegmentState::FirstTimeSent ||
+      segInfo.state == SegmentState::InRetxQueue) { // do not sample RTT for retransmitted segments
+    size_t nExpectedSamples = std::max(static_cast<int>(std::ceil(m_nInFlight / 2.0)), 1);
+    m_rttEstimator.addMeasurement(recvSegNo, rtt, nExpectedSamples);
+    m_segmentInfo.erase(recvSegNo); // remove the entry associated with the received segment
+  }
+  else { // retransmission
+    segInfo.state = SegmentState::RetxReceived;
+  }
+
+  BOOST_ASSERT(m_nReceived > 0);
+  if (m_hasFinalBlockId && m_nReceived - 1 >= m_lastSegmentNo) { // all segments have been received
+    cancel();
+    if (m_options.isVerbose) {
+      printSummary();
+    }
+  }
+  else {
+    schedulePackets();
+  }
+}
+
+void
+PipelineInterestsAimd::handleNack(const Interest& interest, const lp::Nack& nack)
+{
+  if (isStopping())
+    return;
+
+  if (m_options.isVerbose)
+    std::cerr << "Received Nack with reason " << nack.getReason()
+              << " for Interest " << interest << std::endl;
+
+  uint64_t segNo = interest.getName()[-1].toSegment();
+
+  switch (nack.getReason()) {
+    case lp::NackReason::DUPLICATE: {
+      break; // ignore duplicates
+    }
+    case lp::NackReason::CONGESTION: { // treated the same as timeout for now
+      m_retxQueue.push(segNo); // put on retx queue
+      m_segmentInfo[segNo].state = SegmentState::InRetxQueue; // update state
+      handleTimeout(1);
+      break;
+    }
+    default: {
+      handleFail(segNo, "Could not retrieve data for " + interest.getName().toUri() +
+                 ", reason: " + boost::lexical_cast<std::string>(nack.getReason()));
+      break;
+    }
+  }
+}
+
+void
+PipelineInterestsAimd::handleLifetimeExpiration(const Interest& interest)
+{
+  if (isStopping())
+    return;
+
+  uint64_t segNo = interest.getName()[-1].toSegment();
+  m_retxQueue.push(segNo); // put on retx queue
+  m_segmentInfo[segNo].state = SegmentState::InRetxQueue; // update state
+  handleTimeout(1);
+}
+
+void
+PipelineInterestsAimd::handleTimeout(int timeoutCount)
+{
+  if (timeoutCount <= 0)
+    return;
+
+  if (m_options.disableCwa || m_highData > m_recPoint) {
+    // react to only one timeout per RTT (conservative window adaptation)
+    m_recPoint = m_highInterest;
+
+    decreaseWindow();
+    m_rttEstimator.backoffRto();
+    m_nLossEvents++;
+
+    if (m_options.isVerbose) {
+      std::cerr << "Packet loss event, cwnd = " << m_cwnd
+                << ", ssthresh = " << m_ssthresh << std::endl;
+    }
+  }
+
+  if (m_nInFlight > static_cast<uint64_t>(timeoutCount))
+    m_nInFlight -= timeoutCount;
+  else
+    m_nInFlight = 0;
+
+  schedulePackets();
+}
+
+void
+PipelineInterestsAimd::handleFail(uint64_t segNo, const std::string& reason)
+{
+  if (isStopping())
+    return;
+
+  // if the failed segment is definitely part of the content, raise a fatal error
+  if (m_hasFinalBlockId && segNo <= m_lastSegmentNo)
+    return onFailure(reason);
+
+  if (!m_hasFinalBlockId) {
+    m_segmentInfo.erase(segNo);
+    if (m_nInFlight > 0)
+      m_nInFlight--;
+
+    if (m_segmentInfo.empty()) {
+      onFailure("Fetching terminated but no final segment number has been found");
+    }
+    else {
+      cancelInFlightSegmentsGreaterThan(segNo);
+      m_hasFailure = true;
+      m_failedSegNo = segNo;
+      m_failureReason = reason;
+    }
+  }
+}
+
+void
+PipelineInterestsAimd::increaseWindow()
+{
+  if (m_cwnd < m_ssthresh) {
+    m_cwnd += m_options.aiStep; // additive increase
+  } else {
+    m_cwnd += m_options.aiStep / std::floor(m_cwnd); // congestion avoidance
+  }
+  afterCwndChange(time::steady_clock::now() - m_startTime, m_cwnd);
+}
+
+void
+PipelineInterestsAimd::decreaseWindow()
+{
+  // please refer to RFC 5681, Section 3.1 for the rationale behind it
+  m_ssthresh = std::max(2.0, m_cwnd * m_options.mdCoef); // multiplicative decrease
+  m_cwnd = m_options.resetCwndToInit ? m_options.initCwnd : m_ssthresh;
+  afterCwndChange(time::steady_clock::now() - m_startTime, m_cwnd);
+}
+
+uint64_t
+PipelineInterestsAimd::getNextSegmentNo()
+{
+  // get around the excluded segment
+  if (m_nextSegmentNo == m_excludedSegmentNo)
+    m_nextSegmentNo++;
+  return m_nextSegmentNo++;
+}
+
+void
+PipelineInterestsAimd::cancelInFlightSegmentsGreaterThan(uint64_t segmentNo)
+{
+  for (auto it = m_segmentInfo.begin(); it != m_segmentInfo.end();) {
+    // cancel fetching all segments that follow
+    if (it->first > segmentNo) {
+      m_face.removePendingInterest(it->second.interestId);
+      it = m_segmentInfo.erase(it);
+      if (m_nInFlight > 0)
+        m_nInFlight--;
+    }
+    else {
+      ++it;
+    }
+  }
+}
+
+void
+PipelineInterestsAimd::printSummary() const
+{
+  Milliseconds timePassed = time::steady_clock::now() - m_startTime;
+  double throughput = (8 * m_receivedSize * 1000) / timePassed.count();
+
+  int pow = 0;
+  std::string throughputUnit;
+  while (throughput >= 1000.0 && pow < 4) {
+    throughput /= 1000.0;
+    pow++;
+  }
+  switch (pow) {
+    case 0:
+      throughputUnit = "bit/s";
+      break;
+    case 1:
+      throughputUnit = "kbit/s";
+      break;
+    case 2:
+      throughputUnit = "Mbit/s";
+      break;
+    case 3:
+      throughputUnit = "Gbit/s";
+      break;
+    case 4:
+      throughputUnit = "Tbit/s";
+      break;
+  }
+
+  std::cerr << "\nAll segments have been received.\n"
+            << "Total # of segments received: " << m_nReceived << "\n"
+            << "Time used: " << timePassed.count() << " ms" << "\n"
+            << "Total # of packet loss burst: " << m_nLossEvents << "\n"
+            << "Packet loss rate: "
+            << static_cast<double>(m_nLossEvents) / static_cast<double>(m_nReceived) << "\n"
+            << "Total # of retransmitted segments: " << m_nRetransmitted << "\n"
+            << "Goodput: " << throughput << " " << throughputUnit << "\n";
+}
+
+std::ostream&
+operator<<(std::ostream& os, SegmentState state)
+{
+  switch (state) {
+  case SegmentState::FirstTimeSent:
+    os << "FirstTimeSent";
+    break;
+  case SegmentState::InRetxQueue:
+    os << "InRetxQueue";
+    break;
+  case SegmentState::Retransmitted:
+    os << "Retransmitted";
+    break;
+  case SegmentState::RetxReceived:
+    os << "RetxReceived";
+    break;
+  }
+
+  return os;
+}
+
+std::ostream&
+operator<<(std::ostream& os, const PipelineInterestsAimdOptions& options)
+{
+  os << "PipelineInterestsAimd initial parameters:" << "\n"
+     << "\tInitial congestion window size = " << options.initCwnd << "\n"
+     << "\tInitial slow start threshold = " << options.initSsthresh << "\n"
+     << "\tMultiplicative decrease factor = " << options.mdCoef << "\n"
+     << "\tAdditive increase step = " << options.aiStep << "\n"
+     << "\tRTO check interval = " << options.rtoCheckInterval << "\n"
+     << "\tMax retries on timeout or Nack = " << options.maxRetriesOnTimeoutOrNack << "\n";
+
+  std::string cwaStatus = options.disableCwa ? "disabled" : "enabled";
+  os << "\tConservative Window Adaptation " << cwaStatus << "\n";
+
+  std::string cwndStatus = options.resetCwndToInit ? "initCwnd" : "ssthresh";
+  os << "\tResetting cwnd to " << cwndStatus << " when loss event occurs" << "\n";
+  return os;
+}
+
+} // namespace aimd
+} // namespace chunks
+} // namespace ndn

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

@@ -0,0 +1,230 @@
+/**
+ * Copyright (c) 2016,  Regents of the University of California,
+ *                      Colorado State University,
+ *                      University Pierre & Marie Curie, Sorbonne University.
+ *
+ * This file is part of ndn-tools (Named Data Networking Essential Tools).
+ * See AUTHORS.md for complete list of ndn-tools authors and contributors.
+ *
+ * ndn-tools is free software: you can redistribute it and/or modify it under the terms
+ * of the GNU General Public License as published by the Free Software Foundation,
+ * either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-tools is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along with
+ * ndn-tools, e.g., in COPYING.md file.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ *
+ * @author Shuo Yang
+ * @author Weiwei Liu
+ */
+
+#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_AIMD_HPP
+#define NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_AIMD_HPP
+
+#include "options.hpp"
+#include "aimd-rtt-estimator.hpp"
+#include "pipeline-interests.hpp"
+
+#include <queue>
+
+namespace ndn {
+namespace chunks {
+namespace aimd {
+
+struct PipelineInterestsAimdOptions : public Options
+{
+  bool isVerbose = false;
+  double initCwnd = 1.0; ///< initial congestion window size
+  double initSsthresh = std::numeric_limits<double>::max(); ///<  initial slow start threshold
+  double mdCoef = 0.5; ///< multiplicative decrease coefficient
+  double aiStep = 1.0; ///< additive increase step (unit: segment)
+  time::milliseconds rtoCheckInterval = time::milliseconds(10); ///<  time interval for checking retransmission timer
+  bool disableCwa = false; ///< disable Conservative Window Adaptation
+  bool resetCwndToInit = false; ///< reduce cwnd to initCwnd when loss event occurs
+};
+
+/**
+ * @brief indicates the state of the segment
+ */
+enum class SegmentState {
+  FirstTimeSent, ///< segment has been sent for the first time
+  InRetxQueue,   ///< segment is in retransmission queue
+  Retransmitted, ///< segment has been retransmitted
+  RetxReceived,  ///< segment has been received after retransmission
+};
+
+std::ostream&
+operator<<(std::ostream& os, SegmentState state);
+
+/**
+ * @brief Wraps up information that's necessary for segment transmission
+ */
+struct SegmentInfo
+{
+  const PendingInterestId* interestId; ///< The pending interest ID returned by
+                                       ///< ndn::Face::expressInterest. It can be used with
+                                       ///< removePendingInterest before retransmitting this Interest.
+  SegmentState state;
+  Milliseconds rto;
+  time::steady_clock::TimePoint timeSent;
+};
+
+/**
+ * @brief Service for retrieving Data via an Interest pipeline
+ *
+ * Retrieves all segmented Data under the specified prefix by maintaining a dynamic AIMD
+ * congestion window combined with a Conservative Loss Adaptation algorithm. For details,
+ * please refer to the description in section "Interest pipeline types in ndncatchunks" of
+ * tools/chunks/README.md
+ *
+ * Provides retrieved Data on arrival with no ordering guarantees. Data is delivered to the
+ * PipelineInterests' user via callback immediately upon arrival.
+ */
+class PipelineInterestsAimd : public PipelineInterests
+{
+public:
+  typedef PipelineInterestsAimdOptions Options;
+
+public:
+  /**
+   * @brief create a PipelineInterestsAimd service
+   *
+   * Configures the pipelining service without specifying the retrieval namespace. After this
+   * configuration the method run must be called to start the Pipeline.
+   */
+  PipelineInterestsAimd(Face& face, RttEstimator& rttEstimator,
+                        const Options& options = Options());
+
+  ~PipelineInterestsAimd() final;
+
+  /**
+   * @brief Signals when cwnd changes
+   *
+   * The callback function should be: void(Milliseconds age, double cwnd) where age is the
+   * duration since pipeline starts, and cwnd is the new congestion window size (in segments).
+   */
+  signal::Signal<PipelineInterestsAimd, Milliseconds, double> afterCwndChange;
+
+private:
+  /**
+   * @brief fetch all the segments between 0 and lastSegment of the specified prefix
+   *
+   * Starts the pipeline with an AIMD algorithm to control the window size. The pipeline will fetch
+   * every segment until the last segment is successfully received or an error occurs.
+   * The segment with segment number equal to m_excludedSegmentNo will not be fetched.
+   */
+  virtual void
+  doRun() final;
+
+  /**
+   * @brief stop all fetch operations
+   */
+  virtual void
+  doCancel() final;
+
+  /**
+   * @brief check RTO for all sent-but-not-acked segments.
+   */
+  void
+  checkRto();
+
+  /**
+   * @param segNo the segment # of the to-be-sent Interest
+   * @param isRetransmission true if this is a retransmission
+   */
+  void
+  sendInterest(uint64_t segNo, bool isRetransmission);
+
+  void
+  schedulePackets();
+
+  void
+  handleData(const Interest& interest, const Data& data);
+
+  void
+  handleNack(const Interest& interest, const lp::Nack& nack);
+
+  void
+  handleLifetimeExpiration(const Interest& interest);
+
+  void
+  handleTimeout(int timeoutCount);
+
+  void
+  handleFail(uint64_t segNo, const std::string& reason);
+
+  /**
+   * @brief increase congestion window size based on AIMD scheme
+   */
+  void
+  increaseWindow();
+
+  /**
+   * @brief decrease congestion window size based on AIMD scheme
+   */
+  void
+  decreaseWindow();
+
+  /** \return next segment number to retrieve
+   *  \post m_nextSegmentNo == return-value + 1
+   */
+  uint64_t
+  getNextSegmentNo();
+
+  void
+  cancelInFlightSegmentsGreaterThan(uint64_t segmentNo);
+
+  void
+  printSummary() const;
+
+PUBLIC_WITH_TESTS_ELSE_PRIVATE:
+  const Options m_options;
+  RttEstimator& m_rttEstimator;
+  Scheduler m_scheduler;
+  uint64_t m_nextSegmentNo;
+  size_t m_receivedSize;
+
+  uint64_t m_highData; ///< the highest segment number of the Data packet the consumer has received so far
+  uint64_t m_highInterest; ///< the highest segment number of the Interests the consumer has sent so far
+  uint64_t m_recPoint; ///< the value of m_highInterest when a packet loss event occurred
+                       ///< It remains fixed until the next packet loss event happens
+
+  uint64_t m_nInFlight; ///< # of segments in flight
+  uint64_t m_nReceived; ///< # of segments received
+  uint64_t m_nLossEvents; ///< # of loss events occurred
+  uint64_t m_nRetransmitted; ///< # of segments retransmitted
+
+  time::steady_clock::TimePoint m_startTime; ///<  start time of pipelining
+
+  double m_cwnd; ///< current congestion window size (in segments)
+  double m_ssthresh; ///< current slow start threshold
+
+  std::queue<uint64_t> m_retxQueue;
+
+  std::unordered_map<uint64_t, SegmentInfo> m_segmentInfo; ///< the map keeps all the internal information
+                                                           ///< of the sent but not ackownledged segments
+
+  std::unordered_map<uint64_t, int> m_retxCount; ///< maps segment number to its retransmission count.
+                                                 ///< if the count reaches to the maximum number of
+                                                 ///< timeout/nack retries, the pipeline will be aborted
+  bool m_hasFailure;
+  uint64_t m_failedSegNo;
+  std::string m_failureReason;
+};
+
+std::ostream&
+operator<<(std::ostream& os, const PipelineInterestsAimdOptions& options);
+
+} // namespace aimd
+
+using aimd::PipelineInterestsAimd;
+
+} // namespace chunks
+} // namespace ndn
+
+#endif // NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_AIMD_HPP

tools/chunks/catchunks/pipeline-interests.hpp

@@ -119,6 +119,8 @@ protected:
   Name m_prefix;
   uint64_t m_lastSegmentNo;
   uint64_t m_excludedSegmentNo;
+
+PUBLIC_WITH_TESTS_ELSE_PROTECTED:
   bool m_hasFinalBlockId; ///< true if the last segment number is known
 
 private: