chunks: move getNextSegmentNo() to PipelineInterests base class

This hides the "excluded segment" logic from subclasses and
reduces code duplication.

Change-Id: I9f7b5b88e6010e4e16a0876e79fc2ecb9ff8dc00
This commit is contained in:
Davide Pesavento committed 2017-11-04 20:21:38 -04:00
1 parent 4d36ed5889
commit b587cfd779
6 files changed
+45 -51

No files matched your search

@@ -40,7 +40,6 @@ PipelineInterestsAimd::PipelineInterestsAimd(Face& face, RttEstimator& rttEstima
, m_rttEstimator(rttEstimator)
, m_scheduler(m_face.getIoService())
, m_checkRtoEvent(m_scheduler)
, m_nextSegmentNo(0)
, m_highData(0)
, m_highInterest(0)
, m_recPoint(0)
@@ -394,15 +393,6 @@ PipelineInterestsAimd::decreaseWindow()
afterCwndChange(time::steady_clock::now() - getStartTime(), 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 segNo)
{
@@ -122,9 +122,8 @@ 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.
* 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.
*/
void
doRun() final;
@@ -181,12 +180,6 @@ private:
void
decreaseWindow();
/** \return next segment number to retrieve
* \post m_nextSegmentNo == return-value + 1
*/
uint64_t
getNextSegmentNo();
void
cancelInFlightSegmentsGreaterThan(uint64_t segNo);
@@ -199,8 +192,6 @@ PUBLIC_WITH_TESTS_ELSE_PRIVATE:
Scheduler m_scheduler;
scheduler::ScopedEventId m_checkRtoEvent;
uint64_t m_nextSegmentNo;
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,
@@ -36,7 +36,6 @@ namespace chunks {
PipelineInterestsFixedWindow::PipelineInterestsFixedWindow(Face& face, const Options& options)
: PipelineInterests(face)
, m_options(options)
, m_nextSegmentNo(0)
, m_hasFailure(false)
{
m_segmentFetchers.resize(m_options.maxPipelineSize);
@@ -71,17 +70,15 @@ PipelineInterestsFixedWindow::fetchNextSegment(std::size_t pipeNo)
return false;
}
if (m_nextSegmentNo == m_excludedSegmentNo)
m_nextSegmentNo++;
if (m_hasFinalBlockId && m_nextSegmentNo > m_lastSegmentNo)
return false;
uint64_t nextSegmentNo = getNextSegmentNo();
if (m_hasFinalBlockId && nextSegmentNo > m_lastSegmentNo)
return false;
// send interest for next segment
if (m_options.isVerbose)
std::cerr << "Requesting segment #" << m_nextSegmentNo << std::endl;
std::cerr << "Requesting segment #" << nextSegmentNo << std::endl;
Interest interest(Name(m_prefix).appendSegment(m_nextSegmentNo));
Interest interest(Name(m_prefix).appendSegment(nextSegmentNo));
interest.setInterestLifetime(m_options.interestLifetime);
interest.setMustBeFresh(m_options.mustBeFresh);
interest.setMaxSuffixComponents(1);
@@ -95,8 +92,7 @@ PipelineInterestsFixedWindow::fetchNextSegment(std::size_t pipeNo)
m_options.isVerbose);
BOOST_ASSERT(!m_segmentFetchers[pipeNo].first || !m_segmentFetchers[pipeNo].first->isRunning());
m_segmentFetchers[pipeNo] = make_pair(fetcher, m_nextSegmentNo);
m_nextSegmentNo++;
m_segmentFetchers[pipeNo] = make_pair(fetcher, nextSegmentNo);
return true;
}
@@ -85,7 +85,6 @@ private:
*
* Starts a fixed-window pipeline with size equal to m_options.maxPipelineSize. 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.
*/
void
doRun() final;
@@ -110,7 +109,7 @@ private:
private:
const Options m_options;
std::vector<std::pair<shared_ptr<DataFetcher>, uint64_t>> m_segmentFetchers;
uint64_t m_nextSegmentNo;
/**
* 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
+15 -4
View File
@@ -36,10 +36,11 @@ namespace chunks {
PipelineInterests::PipelineInterests(Face& face)
: m_face(face)
, m_lastSegmentNo(0)
, m_excludedSegmentNo(0)
, m_receivedSize(0)
, m_nReceived(0)
, m_receivedSize(0)
, m_hasFinalBlockId(false)
, m_nextSegmentNo(0)
, m_excludedSegmentNo(0)
, m_isStopping(false)
{
}
@@ -78,6 +79,16 @@ PipelineInterests::cancel()
doCancel();
}
uint64_t
PipelineInterests::getNextSegmentNo()
{
// skip the excluded segment
if (m_nextSegmentNo == m_excludedSegmentNo)
m_nextSegmentNo++;
return m_nextSegmentNo++;
}
void
PipelineInterests::onFailure(const std::string& reason)
{
@@ -116,8 +127,8 @@ PipelineInterests::formatThroughput(double throughput)
void
PipelineInterests::printSummary() const
{
typedef time::duration<double, time::milliseconds::period> Milliseconds;
Milliseconds timeElapsed = time::steady_clock::now() - getStartTime();
using namespace ndn::time;
duration<double, milliseconds::period> timeElapsed = steady_clock::now() - getStartTime();
double throughput = (8 * m_receivedSize * 1000) / timeElapsed.count();
std::cerr << "\nAll segments have been received.\n"
+21 -14
View File
@@ -93,6 +93,13 @@ protected:
return m_isStopping;
}
/**
* @return next segment number to retrieve
* @post m_nextSegmentNo == return-value + 1
*/
uint64_t
getNextSegmentNo();
void
onData(const Interest& interest, const Data& data) const
{
@@ -105,27 +112,27 @@ protected:
void
onFailure(const std::string& reason);
/**
* @brief print statistics about this fetching session
*
* Subclasses can override this method to print additional stats or change the summary format
*/
virtual void
printSummary() const;
/**
* @param throughput The throughput in bits/s
*/
static std::string
formatThroughput(double throughput);
/**
* @brief print statistics about quality of packet delivery
*
* This method should be overriden by each pipeline (e.g. AIMD)
*/
virtual void
printSummary() const;
private:
/**
* @brief perform subclass-specific operations to fetch all the segments
*
* When overriding this function, at a minimum, the subclass should implement the retrieving
* of all the segments. Segment m_excludedSegmentNo can be skipped. Subclass must guarantee
* that onData is called at least once for every segment that is fetched successfully.
* of all the segments. Subclass must guarantee that onData is called at least once for every
* segment that is fetched successfully.
*
* @note m_lastSegmentNo contains a valid value only if m_hasFinalBlockId is true.
*/
@@ -138,11 +145,9 @@ private:
protected:
Face& m_face;
Name m_prefix;
uint64_t m_lastSegmentNo;
uint64_t m_excludedSegmentNo;
size_t m_receivedSize; ///< size of received data in bytes
uint64_t m_lastSegmentNo; ///< valid only if m_hasFinalBlockId == true
int64_t m_nReceived; ///< number of segments received
size_t m_receivedSize; ///< size of received data in bytes
PUBLIC_WITH_TESTS_ELSE_PROTECTED:
bool m_hasFinalBlockId; ///< true if the last segment number is known
@@ -150,6 +155,8 @@ PUBLIC_WITH_TESTS_ELSE_PROTECTED:
private:
DataCallback m_onData;
FailureCallback m_onFailure;
uint64_t m_nextSegmentNo;
uint64_t m_excludedSegmentNo;
time::steady_clock::TimePoint m_startTime;
bool m_isStopping;
};