SunnyQjm/ndn-tools
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

catchunks: simplify internal options handling

Browse Source 
Change-Id: I537a281f6c996c2544c145ab9cc01a54272c8efa
This commit is contained in:
Davide Pesavento committed 2019-10-18 12:51:10 -04:00
1 parent 32c0df2e9e
commit 97a33b200d
19 files changed
+169 -303
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/chunks/consumer.t.cpp
+5 -8
View File
@@ -136,11 +136,7 @@ BOOST_AUTO_TEST_CASE(OutputDataUnordered)
class PipelineInterestsDummy : public PipelineInterests
{
public:
PipelineInterestsDummy(Face& face)
: PipelineInterests(face)
, isPipelineRunning(false)
{
}
using PipelineInterests::PipelineInterests;
private:
void
@@ -155,18 +151,19 @@ private:
}
public:
bool isPipelineRunning;
bool isPipelineRunning = false;
};
BOOST_FIXTURE_TEST_CASE(RunBasic, UnitTestTimeFixture)
{
boost::asio::io_service io;
util::DummyClientFace face(io);
Options options;
Consumer consumer(security::v2::getAcceptAllValidator());
Name prefix = Name("/ndn/chunks/test").appendVersion(1);
auto discover = make_unique<DiscoverVersion>(prefix, face, Options());
auto pipeline = make_unique<PipelineInterestsDummy>(face);
auto discover = make_unique<DiscoverVersion>(face, prefix, options);
auto pipeline = make_unique<PipelineInterestsDummy>(face, options);
auto pipelinePtr = pipeline.get();
consumer.run(std::move(discover), std::move(pipeline));
tests/chunks/discover-version.t.cpp
+9 -19
View File
@@ -41,28 +41,18 @@ class DiscoverVersionFixture : public UnitTestTimeFixture,
public IdentityManagementFixture
{
public:
DiscoverVersionFixture()
: face(io)
, name("/ndn/chunks/test")
, isDiscoveryFinished(false)
, version(1449227841747)
{
opt.interestLifetime = ndn::DEFAULT_INTEREST_LIFETIME;
opt.maxRetriesOnTimeoutOrNack = 15;
opt.isVerbose = false;
}
void
run(const Name& prefix)
{
BOOST_REQUIRE(!prefix.empty());
discover = make_unique<DiscoverVersion>(prefix, face, opt);
discover = make_unique<DiscoverVersion>(face, prefix, opt);
discover->onDiscoverySuccess.connect([this] (const Name& versionedName) {
isDiscoveryFinished = true;
BOOST_REQUIRE(!versionedName.empty() && versionedName[-1].isVersion());
discoveredVersion = versionedName[-1].toVersion();
isDiscoveryFinished = true;
});
discover->onDiscoveryFailure.connect([this] (const std::string& reason) {
discover->onDiscoveryFailure.connect([this] (const std::string&) {
isDiscoveryFinished = true;
});
@@ -72,13 +62,13 @@ public:
protected:
boost::asio::io_service io;
util::DummyClientFace face;
Name name;
util::DummyClientFace face{io};
Name name = "/ndn/chunks/test";
Options opt;
unique_ptr<DiscoverVersion> discover;
optional<uint64_t> discoveredVersion;
bool isDiscoveryFinished;
uint64_t version;
Options opt;
bool isDiscoveryFinished = false;
uint64_t version = 1449227841747;
};
BOOST_AUTO_TEST_SUITE(Chunks)
tests/chunks/pipeline-interests-aimd.t.cpp
+3 -20
View File
@@ -26,7 +26,6 @@
*/
#include "tools/chunks/catchunks/pipeline-interests-aimd.hpp"
#include "tools/chunks/catchunks/options.hpp"
#include "pipeline-interests-fixture.hpp"
@@ -40,9 +39,9 @@ class PipelineInterestAimdFixture : public PipelineInterestsFixture
{
public:
PipelineInterestAimdFixture()
: opt(makePipelineOptions())
, rttEstimator(makeRttEstimatorOptions())
: rttEstimator(makeRttEstimatorOptions())
{
opt.isQuiet = true;
createPipeline();
}
@@ -55,22 +54,6 @@ public:
}
private:
static PipelineInterestsAdaptive::Options
makePipelineOptions()
{
PipelineInterestsAdaptive::Options pipelineOptions;
pipelineOptions.isQuiet = true;
pipelineOptions.isVerbose = false;
pipelineOptions.disableCwa = false;
pipelineOptions.ignoreCongMarks = 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 shared_ptr<RttEstimatorWithStats::Options>
makeRttEstimatorOptions()
{
@@ -86,7 +69,7 @@ private:
}
protected:
PipelineInterestsAdaptive::Options opt;
Options opt;
RttEstimatorWithStats rttEstimator;
PipelineInterestsAdaptive* pipeline;
static constexpr double MARGIN = 0.001;
tests/chunks/pipeline-interests-cubic.t.cpp
+3 -22
View File
@@ -26,7 +26,6 @@
*/
#include "tools/chunks/catchunks/pipeline-interests-cubic.hpp"
#include "tools/chunks/catchunks/options.hpp"
#include "pipeline-interests-fixture.hpp"
@@ -40,9 +39,9 @@ class PipelineInterestCubicFixture : public PipelineInterestsFixture
{
public:
PipelineInterestCubicFixture()
: opt(makePipelineOptions())
, rttEstimator(makeRttEstimatorOptions())
: rttEstimator(makeRttEstimatorOptions())
{
opt.isQuiet = true;
createPipeline();
}
@@ -55,24 +54,6 @@ public:
}
private:
static PipelineInterestsCubic::Options
makePipelineOptions()
{
PipelineInterestsCubic::Options pipelineOptions;
pipelineOptions.isQuiet = true;
pipelineOptions.isVerbose = false;
pipelineOptions.disableCwa = false;
pipelineOptions.ignoreCongMarks = false;
pipelineOptions.resetCwndToInit = false;
pipelineOptions.initCwnd = 1.0;
pipelineOptions.aiStep = 1.0;
pipelineOptions.mdCoef = 0.5;
pipelineOptions.initSsthresh = std::numeric_limits<int>::max();
pipelineOptions.cubicBeta = 0.7;
pipelineOptions.enableFastConv = false;
return pipelineOptions;
}
static shared_ptr<RttEstimatorWithStats::Options>
makeRttEstimatorOptions()
{
@@ -88,7 +69,7 @@ private:
}
protected:
PipelineInterestsCubic::Options opt;
Options opt;
RttEstimatorWithStats rttEstimator;
PipelineInterestsCubic* pipeline;
static constexpr double MARGIN = 0.001;
tests/chunks/pipeline-interests-fixed.t.cpp
+5 -15
View File
@@ -37,8 +37,11 @@ class PipelineInterestFixedFixture : public PipelineInterestsFixture
{
public:
PipelineInterestFixedFixture()
: opt(makeOptions())
{
opt.interestLifetime = 1_s;
opt.maxRetriesOnTimeoutOrNack = 3;
opt.isQuiet = true;
opt.maxPipelineSize = 5;
createPipeline();
}
@@ -50,21 +53,8 @@ public:
setPipeline(std::move(pline));
}
private:
static PipelineInterestsFixed::Options
makeOptions()
{
PipelineInterestsFixed::Options options;
options.isQuiet = true;
options.isVerbose = false;
options.interestLifetime = time::seconds(1);
options.maxRetriesOnTimeoutOrNack = 3;
options.maxPipelineSize = 5;
return options;
}
protected:
PipelineInterestsFixed::Options opt;
Options opt;
PipelineInterestsFixed* pipeline;
};
tools/chunks/catchunks/discover-version.cpp
+10 -10
View File
@@ -34,10 +34,10 @@
namespace ndn {
namespace chunks {
DiscoverVersion::DiscoverVersion(const Name& prefix, Face& face, const Options& options)
: chunks::Options(options)
DiscoverVersion::DiscoverVersion(Face& face, const Name& prefix, const Options& options)
: m_face(face)
, m_prefix(prefix)
, m_face(face)
, m_options(options)
{
}
@@ -50,24 +50,24 @@ DiscoverVersion::run()
}
Interest interest = MetadataObject::makeDiscoveryInterest(m_prefix)
.setInterestLifetime(interestLifetime);
.setInterestLifetime(m_options.interestLifetime);
m_fetcher = DataFetcher::fetch(m_face, interest,
maxRetriesOnTimeoutOrNack, maxRetriesOnTimeoutOrNack,
m_options.maxRetriesOnTimeoutOrNack, m_options.maxRetriesOnTimeoutOrNack,
bind(&DiscoverVersion::handleData, this, _1, _2),
[this] (const Interest& interest, const std::string& reason) {
[this] (const Interest&, const std::string& reason) {
onDiscoveryFailure(reason);
},
[this] (const Interest& interest, const std::string& reason) {
[this] (const Interest&, const std::string& reason) {
onDiscoveryFailure(reason);
},
isVerbose);
m_options.isVerbose);
}
void
DiscoverVersion::handleData(const Interest& interest, const Data& data)
{
if (isVerbose)
if (m_options.isVerbose)
std::cerr << "Data: " << data << std::endl;
// make a metadata object from received metadata packet
@@ -85,7 +85,7 @@ DiscoverVersion::handleData(const Interest& interest, const Data& data)
return;
}
if (isVerbose) {
if (m_options.isVerbose) {
std::cerr << "Discovered Data version: " << mobject.getVersionedName()[-1] << std::endl;
}
tools/chunks/catchunks/discover-version.hpp
+5 -8
View File
@@ -29,7 +29,6 @@
#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_DISCOVER_VERSION_HPP
#define NDN_TOOLS_CHUNKS_CATCHUNKS_DISCOVER_VERSION_HPP
#include "core/common.hpp"
#include "options.hpp"
namespace ndn {
@@ -38,12 +37,12 @@ namespace chunks {
class DataFetcher;
/**
* @brief Base class of services for discovering the latest Data version
* @brief Service for discovering the latest Data version.
*
* DiscoverVersion's user is notified once after identifying the latest retrievable version or
* on failure to find any Data version.
*/
class DiscoverVersion : virtual protected Options, noncopyable
class DiscoverVersion
{
public: // signals
/**
@@ -57,10 +56,7 @@ public: // signals
signal::Signal<DiscoverVersion, std::string> onDiscoveryFailure;
public:
/**
* @brief create a DiscoverVersion service
*/
DiscoverVersion(const Name& prefix, Face& face, const Options& options);
DiscoverVersion(Face& face, const Name& prefix, const Options& options);
/**
* @brief identify the latest Data version published.
@@ -73,8 +69,9 @@ private:
handleData(const Interest& interest, const Data& data);
private:
const Name m_prefix;
Face& m_face;
const Name m_prefix;
const Options& m_options;
shared_ptr<DataFetcher> m_fetcher;
};
tools/chunks/catchunks/main.cpp
+21 -43
View File
@@ -31,7 +31,6 @@
#include "consumer.hpp"
#include "discover-version.hpp"
#include "options.hpp"
#include "pipeline-interests-aimd.hpp"
#include "pipeline-interests-cubic.hpp"
#include "pipeline-interests-fixed.hpp"
@@ -48,18 +47,12 @@ static int
main(int argc, char* argv[])
{
std::string programName(argv[0]);
Options options;
std::string pipelineType("cubic");
size_t maxPipelineSize(1);
std::string uri;
// congestion control parameters
bool disableCwa(false), resetCwndToInit(false),
ignoreCongMarks(false), enableFastConv(false);
int initCwnd(1), initSsthresh(std::numeric_limits<int>::max()), k(8);
double aiStep(1.0), rtoAlpha(0.125), rtoBeta(0.25), aimdBeta(0.5), cubicBeta(0.7);
Options options;
std::string uri, pipelineType("cubic"), cwndPath, rttPath;
time::milliseconds::rep minRto(200), maxRto(60000);
std::string cwndPath, rttPath;
double rtoAlpha(0.125), rtoBeta(0.25);
int rtoK(8);
namespace po = boost::program_options;
po::options_description basicDesc("Basic Options");
@@ -79,33 +72,33 @@ main(int argc, char* argv[])
po::options_description fixedPipeDesc("Fixed pipeline options");
fixedPipeDesc.add_options()
("pipeline-size,s", po::value<size_t>(&maxPipelineSize)->default_value(maxPipelineSize),
("pipeline-size,s", po::value<size_t>(&options.maxPipelineSize)->default_value(options.maxPipelineSize),
"size of the Interest pipeline")
;
po::options_description adaptivePipeDesc("Adaptive pipeline options (AIMD & CUBIC)");
adaptivePipeDesc.add_options()
("ignore-marks", po::bool_switch(&ignoreCongMarks),
("ignore-marks", po::bool_switch(&options.ignoreCongMarks),
"do not decrease the window after receiving a congestion mark")
("disable-cwa", po::bool_switch(&disableCwa),
("disable-cwa", po::bool_switch(&options.disableCwa),
"disable Conservative Window Adaptation, i.e., reduce the window on "
"each timeout or congestion mark instead of at most once per RTT")
("reset-cwnd-to-init", po::bool_switch(&resetCwndToInit),
("reset-cwnd-to-init", po::bool_switch(&options.resetCwndToInit),
"after a timeout or congestion mark, reset the window "
"to the initial value instead of resetting to ssthresh")
("init-cwnd", po::value<int>(&initCwnd)->default_value(initCwnd),
("init-cwnd", po::value<double>(&options.initCwnd)->default_value(options.initCwnd),
"initial congestion window in segments")
("init-ssthresh", po::value<int>(&initSsthresh),
("init-ssthresh", po::value<double>(&options.initSsthresh),
"initial slow start threshold in segments (defaults to infinity)")
("aimd-step", po::value<double>(&aiStep)->default_value(aiStep),
("aimd-step", po::value<double>(&options.aiStep)->default_value(options.aiStep),
"additive-increase step")
("aimd-beta", po::value<double>(&aimdBeta)->default_value(aimdBeta),
("aimd-beta", po::value<double>(&options.mdCoef)->default_value(options.mdCoef),
"multiplicative decrease factor (AIMD)")
("rto-alpha", po::value<double>(&rtoAlpha)->default_value(rtoAlpha),
"alpha value for RTO calculation")
("rto-beta", po::value<double>(&rtoBeta)->default_value(rtoBeta),
"beta value for RTO calculation")
("rto-k", po::value<int>(&k)->default_value(k),
("rto-k", po::value<int>(&rtoK)->default_value(rtoK),
"k value for RTO calculation")
("min-rto", po::value<time::milliseconds::rep>(&minRto)->default_value(minRto),
"minimum RTO value, in milliseconds")
@@ -117,9 +110,8 @@ main(int argc, char* argv[])
po::options_description cubicPipeDesc("CUBIC pipeline options");
cubicPipeDesc.add_options()
("fast-conv", po::bool_switch(&enableFastConv), "enable cubic fast convergence")
("cubic-beta", po::value<double>(&cubicBeta),
"window decrease factor for CUBIC (defaults to 0.7)")
("cubic-beta", po::value<double>(&options.cubicBeta), "window decrease factor (defaults to 0.7)")
("fast-conv", po::bool_switch(&options.enableFastConv), "enable fast convergence")
;
po::options_description visibleDesc;
@@ -169,7 +161,7 @@ main(int argc, char* argv[])
return 2;
}
if (maxPipelineSize < 1 || maxPipelineSize > 1024) {
if (options.maxPipelineSize < 1 || options.maxPipelineSize > 1024) {
std::cerr << "ERROR: pipeline size must be between 1 and 1024" << std::endl;
return 2;
}
@@ -192,7 +184,7 @@ main(int argc, char* argv[])
try {
Face face;
auto discover = make_unique<DiscoverVersion>(Name(uri), face, options);
auto discover = make_unique<DiscoverVersion>(face, Name(uri), options);
unique_ptr<PipelineInterests> pipeline;
unique_ptr<StatisticsCollector> statsCollector;
unique_ptr<RttEstimatorWithStats> rttEstimator;
@@ -200,15 +192,13 @@ main(int argc, char* argv[])
std::ofstream statsFileRtt;
if (pipelineType == "fixed") {
PipelineInterestsFixed::Options optionsPipeline(options);
optionsPipeline.maxPipelineSize = maxPipelineSize;
pipeline = make_unique<PipelineInterestsFixed>(face, optionsPipeline);
pipeline = make_unique<PipelineInterestsFixed>(face, options);
}
else if (pipelineType == "aimd" || pipelineType == "cubic") {
auto optionsRttEst = make_shared<RttEstimatorWithStats::Options>();
optionsRttEst->alpha = rtoAlpha;
optionsRttEst->beta = rtoBeta;
optionsRttEst->k = k;
optionsRttEst->k = rtoK;
optionsRttEst->initialRto = 1_s;
optionsRttEst->minRto = time::milliseconds(minRto);
optionsRttEst->maxRto = time::milliseconds(maxRto);
@@ -226,24 +216,12 @@ main(int argc, char* argv[])
}
rttEstimator = make_unique<RttEstimatorWithStats>(std::move(optionsRttEst));
PipelineInterestsAdaptive::Options optionsPipeline(options);
optionsPipeline.disableCwa = disableCwa;
optionsPipeline.resetCwndToInit = resetCwndToInit;
optionsPipeline.initCwnd = initCwnd;
optionsPipeline.initSsthresh = initSsthresh;
optionsPipeline.aiStep = aiStep;
optionsPipeline.mdCoef = aimdBeta;
optionsPipeline.ignoreCongMarks = ignoreCongMarks;
unique_ptr<PipelineInterestsAdaptive> adaptivePipeline;
if (pipelineType == "aimd") {
adaptivePipeline = make_unique<PipelineInterestsAimd>(face, *rttEstimator, optionsPipeline);
adaptivePipeline = make_unique<PipelineInterestsAimd>(face, *rttEstimator, options);
}
else {
PipelineInterestsCubic::Options optionsCubic(optionsPipeline);
optionsCubic.enableFastConv = enableFastConv;
optionsCubic.cubicBeta = cubicBeta;
adaptivePipeline = make_unique<PipelineInterestsCubic>(face, *rttEstimator, optionsCubic);
adaptivePipeline = make_unique<PipelineInterestsCubic>(face, *rttEstimator, options);
}
if (!cwndPath.empty() || !rttPath.empty()) {
tools/chunks/catchunks/options.hpp
+22 -3
View File
@@ -27,19 +27,38 @@
#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_OPTIONS_HPP
#define NDN_TOOLS_CHUNKS_CATCHUNKS_OPTIONS_HPP
#include <ndn-cxx/interest.hpp>
#include <ndn-cxx/util/time.hpp>
#include "core/common.hpp"
namespace ndn {
namespace chunks {
struct Options
struct Options : noncopyable
{
// Common options
time::milliseconds interestLifetime = DEFAULT_INTEREST_LIFETIME;
int maxRetriesOnTimeoutOrNack = 15;
bool mustBeFresh = false;
bool isQuiet = false;
bool isVerbose = false;
// Fixed pipeline options
size_t maxPipelineSize = 1;
// Adaptive pipeline common options
double initCwnd = 1.0; ///< initial congestion window size
double initSsthresh = std::numeric_limits<double>::max(); ///< initial slow start threshold
time::milliseconds rtoCheckInterval{10}; ///< interval for checking retransmission timer
bool ignoreCongMarks = false; ///< disable window decrease after receiving congestion mark
bool disableCwa = false; ///< disable conservative window adaptation
// AIMD pipeline options
double aiStep = 1.0; ///< AIMD additive increase step (in segments)
double mdCoef = 0.5; ///< AIMD multiplicative decrease factor
bool resetCwndToInit = false; ///< reduce cwnd to initCwnd when loss event occurs
// Cubic pipeline options
double cubicBeta = 0.7; ///< cubic multiplicative decrease factor
bool enableFastConv = false; ///< use cubic fast convergence
};
} // namespace chunks
tools/chunks/catchunks/pipeline-interests-adaptive.cpp
+18 -20
View File
@@ -39,9 +39,8 @@ constexpr double PipelineInterestsAdaptive::MIN_SSTHRESH;
PipelineInterestsAdaptive::PipelineInterestsAdaptive(Face& face,
RttEstimatorWithStats& rttEstimator,
const Options& options)
: PipelineInterests(face)
, m_options(options)
const Options& opts)
: PipelineInterests(face, opts)
, m_cwnd(m_options.initCwnd)
, m_ssthresh(m_options.initSsthresh)
, m_rttEstimator(rttEstimator)
@@ -416,6 +415,22 @@ PipelineInterestsAdaptive::cancelInFlightSegmentsGreaterThan(uint64_t segNo)
}
}
void
PipelineInterestsAdaptive::printOptions() const
{
PipelineInterests::printOptions();
std::cerr
<< "\tInitial congestion window size = " << m_options.initCwnd << "\n"
<< "\tInitial slow start threshold = " << m_options.initSsthresh << "\n"
<< "\tAdditive increase step = " << m_options.aiStep << "\n"
<< "\tMultiplicative decrease factor = " << m_options.mdCoef << "\n"
<< "\tRTO check interval = " << m_options.rtoCheckInterval << "\n"
<< "\tReact to congestion marks = " << (m_options.ignoreCongMarks ? "no" : "yes") << "\n"
<< "\tConservative window adaptation = " << (m_options.disableCwa ? "no" : "yes") << "\n"
<< "\tResetting window to " << (m_options.resetCwndToInit ?
"initial value" : "ssthresh") << " upon loss event\n";
}
void
PipelineInterestsAdaptive::printSummary() const
{
@@ -456,22 +471,5 @@ operator<<(std::ostream& os, SegmentState state)
return os;
}
std::ostream&
operator<<(std::ostream& os, const PipelineInterestsAdaptiveOptions& options)
{
os << "Adaptive pipeline parameters:\n"
<< "\tInitial congestion window size = " << options.initCwnd << "\n"
<< "\tInitial slow start threshold = " << options.initSsthresh << "\n"
<< "\tAdditive increase step = " << options.aiStep << "\n"
<< "\tMultiplicative decrease factor = " << options.mdCoef << "\n"
<< "\tRTO check interval = " << options.rtoCheckInterval << "\n"
<< "\tMax retries on timeout or Nack = " << (options.maxRetriesOnTimeoutOrNack == DataFetcher::MAX_RETRIES_INFINITE ?
"infinite" : to_string(options.maxRetriesOnTimeoutOrNack)) << "\n"
<< "\tReact to congestion marks = " << (options.ignoreCongMarks ? "no" : "yes") << "\n"
<< "\tConservative window adaptation = " << (options.disableCwa ? "no" : "yes") << "\n"
<< "\tResetting window to " << (options.resetCwndToInit ? "initCwnd" : "ssthresh") << " upon loss event\n";
return os;
}
} // namespace chunks
} // namespace ndn
tools/chunks/catchunks/pipeline-interests-adaptive.hpp
+4 -30
View File
@@ -29,7 +29,6 @@
#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_ADAPTIVE_HPP
#define NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_ADAPTIVE_HPP
#include "options.hpp"
#include "pipeline-interests.hpp"
#include <ndn-cxx/util/rtt-estimator.hpp>
@@ -42,29 +41,6 @@ namespace chunks {
using util::RttEstimatorWithStats;
class PipelineInterestsAdaptiveOptions : public Options
{
public:
explicit
PipelineInterestsAdaptiveOptions(const Options& options = Options())
: Options(options)
{
}
public:
double initCwnd = 1.0; ///< initial congestion window size
double initSsthresh = std::numeric_limits<double>::max(); ///< initial slow start threshold
double aiStep = 1.0; ///< additive increase step (in segments)
double mdCoef = 0.5; ///< multiplicative decrease coefficient
time::milliseconds rtoCheckInterval{10}; ///< interval for checking retransmission timer
bool disableCwa = false; ///< disable Conservative Window Adaptation
bool resetCwndToInit = false; ///< reduce cwnd to initCwnd when loss event occurs
bool ignoreCongMarks = false; ///< disable window decrease after congestion marks
};
std::ostream&
operator<<(std::ostream& os, const PipelineInterestsAdaptiveOptions& options);
/**
* @brief indicates the state of the segment
*/
@@ -101,9 +77,6 @@ struct SegmentInfo
*/
class PipelineInterestsAdaptive : public PipelineInterests
{
public:
using Options = PipelineInterestsAdaptiveOptions;
public:
/**
* @brief Constructor.
@@ -111,8 +84,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, RttEstimatorWithStats& rttEstimator,
const Options& options = Options());
PipelineInterestsAdaptive(Face& face, RttEstimatorWithStats& rttEstimator, const Options& opts);
~PipelineInterestsAdaptive() override;
@@ -141,6 +113,9 @@ public:
protected:
DECLARE_SIGNAL_EMIT(afterCwndChange)
void
printOptions() const;
private:
/**
* @brief Increase congestion window.
@@ -214,7 +189,6 @@ PUBLIC_WITH_TESTS_ELSE_PRIVATE:
PUBLIC_WITH_TESTS_ELSE_PROTECTED:
static constexpr double MIN_SSTHRESH = 2.0;
const Options m_options;
double m_cwnd; ///< current congestion window size (in segments)
double m_ssthresh; ///< current slow start threshold
tools/chunks/catchunks/pipeline-interests-aimd.cpp
+4 -4
View File
@@ -34,11 +34,11 @@ namespace ndn {
namespace chunks {
PipelineInterestsAimd::PipelineInterestsAimd(Face& face, RttEstimatorWithStats& rttEstimator,
const Options& options)
: PipelineInterestsAdaptive(face, rttEstimator, options)
const Options& opts)
: PipelineInterestsAdaptive(face, rttEstimator, opts)
{
if (options.isVerbose) {
std::cerr << options;
if (m_options.isVerbose) {
printOptions();
}
}
tools/chunks/catchunks/pipeline-interests-aimd.hpp
+1 -2
View File
@@ -40,8 +40,7 @@ namespace chunks {
class PipelineInterestsAimd final : public PipelineInterestsAdaptive
{
public:
PipelineInterestsAimd(Face& face, RttEstimatorWithStats& rttEstimator,
const Options& options = Options());
PipelineInterestsAimd(Face& face, RttEstimatorWithStats& rttEstimator, const Options& opts);
private:
void
tools/chunks/catchunks/pipeline-interests-cubic.cpp
+10 -19
View File
@@ -33,13 +33,14 @@ namespace chunks {
constexpr double CUBIC_C = 0.4;
PipelineInterestsCubic::PipelineInterestsCubic(Face& face, RttEstimatorWithStats& rttEstimator,
const Options& options)
: PipelineInterestsAdaptive(face, rttEstimator, options)
, m_cubicOptions(options)
const Options& opts)
: PipelineInterestsAdaptive(face, rttEstimator, opts)
, m_lastDecrease(time::steady_clock::now())
{
if (options.isVerbose) {
std::cerr << options;
if (m_options.isVerbose) {
printOptions();
std::cerr << "\tCubic beta = " << m_options.cubicBeta << "\n"
<< "\tFast convergence = " << (m_options.enableFastConv ? "yes" : "no") << "\n";
}
}
@@ -64,7 +65,7 @@ PipelineInterestsCubic::increaseWindow()
// 2. Time it takes to increase the window to m_wmax = the cwnd right before the last
// window decrease.
// K = cubic_root(wmax*(1-beta_cubic)/C) (Eq. 2)
const double k = std::cbrt(m_wmax * (1 - m_cubicOptions.cubicBeta) / CUBIC_C);
const double k = std::cbrt(m_wmax * (1 - m_options.cubicBeta) / CUBIC_C);
// 3. Target: W_cubic(t) = C*(t-K)^3 + wmax (Eq. 1)
const double wCubic = CUBIC_C * std::pow(t - k, 3) + m_wmax;
@@ -93,9 +94,9 @@ PipelineInterestsCubic::decreaseWindow()
// before it updates wmax for the current congestion event.
// Current wmax < last_wmax
if (m_cubicOptions.enableFastConv && m_cwnd < m_lastWmax) {
if (m_options.enableFastConv && m_cwnd < m_lastWmax) {
m_lastWmax = m_cwnd;
m_wmax = m_cwnd * (1.0 + m_cubicOptions.cubicBeta) / 2.0;
m_wmax = m_cwnd * (1.0 + m_options.cubicBeta) / 2.0;
}
else {
// Save old cwnd as wmax
@@ -103,22 +104,12 @@ PipelineInterestsCubic::decreaseWindow()
m_wmax = m_cwnd;
}
m_ssthresh = std::max(m_options.initCwnd, m_cwnd * m_cubicOptions.cubicBeta);
m_ssthresh = std::max(m_options.initCwnd, m_cwnd * m_options.cubicBeta);
m_cwnd = m_ssthresh;
m_lastDecrease = time::steady_clock::now();
emitSignal(afterCwndChange, time::steady_clock::now() - getStartTime(), m_cwnd);
}
std::ostream&
operator<<(std::ostream& os, const PipelineInterestsCubicOptions& options)
{
os << static_cast<const PipelineInterestsAdaptiveOptions&>(options)
<< "Cubic pipeline parameters:\n"
<< "\tFast convergence = " << (options.enableFastConv ? "yes" : "no") << "\n"
<< "\tCubic beta = " << options.cubicBeta << "\n";
return os;
}
} // namespace chunks
} // namespace ndn
tools/chunks/catchunks/pipeline-interests-cubic.hpp
+1 -25
View File
@@ -31,24 +31,6 @@
namespace ndn {
namespace chunks {
class PipelineInterestsCubicOptions : public PipelineInterestsAdaptiveOptions
{
public:
explicit
PipelineInterestsCubicOptions(const PipelineInterestsAdaptiveOptions& options =
PipelineInterestsAdaptiveOptions())
: PipelineInterestsAdaptiveOptions(options)
{
}
public:
bool enableFastConv = false; ///< use cubic fast convergence
double cubicBeta = 0.7; ///< multiplicative decrease factor (from Linux kernel: 717/1024)
};
std::ostream&
operator<<(std::ostream& os, const PipelineInterestsCubicOptions& options);
/**
* @brief Implements Cubic window increase and decrease.
*
@@ -58,11 +40,7 @@ operator<<(std::ostream& os, const PipelineInterestsCubicOptions& options);
class PipelineInterestsCubic final : public PipelineInterestsAdaptive
{
public:
using Options = PipelineInterestsCubicOptions;
public:
PipelineInterestsCubic(Face& face, RttEstimatorWithStats& rttEstimator,
const Options& options = Options());
PipelineInterestsCubic(Face& face, RttEstimatorWithStats& rttEstimator, const Options& opts);
private:
void
@@ -72,8 +50,6 @@ private:
decreaseWindow() final;
private:
const Options m_cubicOptions;
double m_wmax = 0.0; ///< window size before last window decrease
double m_lastWmax = 0.0; ///< last wmax
time::steady_clock::TimePoint m_lastDecrease; ///< time of last window decrease
tools/chunks/catchunks/pipeline-interests-fixed.cpp
+7 -4
View File
@@ -33,12 +33,15 @@
namespace ndn {
namespace chunks {
PipelineInterestsFixed::PipelineInterestsFixed(Face& face, const Options& options)
: PipelineInterests(face)
, m_options(options)
, m_hasFailure(false)
PipelineInterestsFixed::PipelineInterestsFixed(Face& face, const Options& opts)
: PipelineInterests(face, opts)
{
m_segmentFetchers.resize(m_options.maxPipelineSize);
if (m_options.isVerbose) {
printOptions();
std::cerr << "\tPipeline size = " << m_options.maxPipelineSize << "\n";
}
}
PipelineInterestsFixed::~PipelineInterestsFixed()
tools/chunks/catchunks/pipeline-interests-fixed.hpp
+2 -28
View File
@@ -30,7 +30,6 @@
#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_FIXED_HPP
#define NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_FIXED_HPP
#include "options.hpp"
#include "pipeline-interests.hpp"
namespace ndn {
@@ -38,20 +37,6 @@ namespace chunks {
class DataFetcher;
class PipelineInterestsFixedOptions : public Options
{
public:
explicit
PipelineInterestsFixedOptions(const Options& options = Options())
: Options(options)
, maxPipelineSize(1)
{
}
public:
size_t maxPipelineSize;
};
/**
* @brief Service for retrieving Data via an Interest pipeline
*
@@ -65,17 +50,7 @@ public:
class PipelineInterestsFixed final : public PipelineInterests
{
public:
typedef PipelineInterestsFixedOptions Options;
public:
/**
* @brief create a PipelineInterestsFixed service
*
* Configures the pipelining service without specifying the retrieval namespace. After this
* configuration the method run must be called to start the Pipeline.
*/
explicit
PipelineInterestsFixed(Face& face, const Options& options = Options());
PipelineInterestsFixed(Face& face, const Options& opts);
~PipelineInterestsFixed() final;
@@ -107,14 +82,13 @@ private:
handleFail(const std::string& reason, size_t pipeNo);
private:
const Options m_options;
std::vector<std::pair<shared_ptr<DataFetcher>, uint64_t>> m_segmentFetchers;
/**
* true if one or more segment fetchers encountered an error; if m_hasFinalBlockId
* is false, this is usually not a fatal error for the pipeline
*/
bool m_hasFailure;
bool m_hasFailure = false;
};
} // namespace chunks
tools/chunks/catchunks/pipeline-interests.cpp
+29 -16
View File
@@ -29,12 +29,14 @@
*/
#include "pipeline-interests.hpp"
#include "data-fetcher.hpp"
namespace ndn {
namespace chunks {
PipelineInterests::PipelineInterests(Face& face)
: m_face(face)
PipelineInterests::PipelineInterests(Face& face, const Options& opts)
: m_options(opts)
, m_face(face)
, m_hasFinalBlockId(false)
, m_lastSegmentNo(0)
, m_nReceived(0)
@@ -106,6 +108,31 @@ PipelineInterests::onFailure(const std::string& reason)
m_face.getIoService().post([this, reason] { m_onFailure(reason); });
}
void
PipelineInterests::printOptions() const
{
std::cerr << "Pipeline parameters:\n"
<< "\tRequest fresh content = " << (m_options.mustBeFresh ? "yes" : "no") << "\n"
<< "\tInterest lifetime = " << m_options.interestLifetime << "\n"
<< "\tMax retries on timeout or Nack = " <<
(m_options.maxRetriesOnTimeoutOrNack == DataFetcher::MAX_RETRIES_INFINITE ?
"infinite" : to_string(m_options.maxRetriesOnTimeoutOrNack)) << "\n";
}
void
PipelineInterests::printSummary() const
{
using namespace ndn::time;
duration<double, seconds::period> timeElapsed = steady_clock::now() - getStartTime();
double throughput = 8 * m_receivedSize / timeElapsed.count();
std::cerr << "\n\nAll segments have been received.\n"
<< "Time elapsed: " << timeElapsed << "\n"
<< "Segments received: " << m_nReceived << "\n"
<< "Transferred size: " << m_receivedSize / 1e3 << " kB" << "\n"
<< "Goodput: " << formatThroughput(throughput) << "\n";
}
std::string
PipelineInterests::formatThroughput(double throughput)
{
@@ -129,19 +156,5 @@ PipelineInterests::formatThroughput(double throughput)
return "";
}
void
PipelineInterests::printSummary() const
{
using namespace ndn::time;
duration<double, seconds::period> timeElapsed = steady_clock::now() - getStartTime();
double throughput = 8 * m_receivedSize / timeElapsed.count();
std::cerr << "\n\nAll segments have been received.\n"
<< "Time elapsed: " << timeElapsed << "\n"
<< "Segments received: " << m_nReceived << "\n"
<< "Transferred size: " << m_receivedSize / 1e3 << " kB" << "\n"
<< "Goodput: " << formatThroughput(throughput) << "\n";
}
} // namespace chunks
} // namespace ndn
tools/chunks/catchunks/pipeline-interests.hpp
+10 -7
View File
@@ -31,7 +31,7 @@
#ifndef NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_HPP
#define NDN_TOOLS_CHUNKS_CATCHUNKS_PIPELINE_INTERESTS_HPP
#include "core/common.hpp"
#include "options.hpp"
namespace ndn {
namespace chunks {
@@ -49,21 +49,20 @@ namespace chunks {
class PipelineInterests
{
public:
using DataCallback = std::function<void(const Data&)>;
using FailureCallback = std::function<void(const std::string& reason)>;
/**
* @brief create a PipelineInterests service
* @brief Constructor.
*
* Configures the pipelining service without specifying the retrieval namespace.
* After construction, the method run() must be called in order to start the pipeline.
*/
explicit
PipelineInterests(Face& face);
PipelineInterests(Face& face, const Options& opts);
virtual
~PipelineInterests();
using DataCallback = std::function<void(const Data&)>;
using FailureCallback = std::function<void(const std::string& reason)>;
/**
* @brief start fetching all the segments of the specified prefix
*
@@ -119,6 +118,9 @@ protected:
void
onFailure(const std::string& reason);
void
printOptions() const;
/**
* @brief print statistics about this fetching session
*
@@ -150,6 +152,7 @@ private:
doCancel() = 0;
protected:
const Options& m_options;
Face& m_face;
Name m_prefix;