chunks: modularize pipeline implementation
Change-Id: Iaa5acbb02a583858db6204716947cceeae793a65
Refs: #3636
diff --git a/tools/chunks/catchunks/pipeline-interests.cpp b/tools/chunks/catchunks/pipeline-interests.cpp
index b07c959..ff817e4 100644
--- a/tools/chunks/catchunks/pipeline-interests.cpp
+++ b/tools/chunks/catchunks/pipeline-interests.cpp
@@ -23,179 +23,63 @@
* @author Wentao Shang
* @author Steve DiBenedetto
* @author Andrea Tosatto
+ * @author Davide Pesavento
+ * @author Weiwei Liu
*/
#include "pipeline-interests.hpp"
-#include "data-fetcher.hpp"
namespace ndn {
namespace chunks {
-PipelineInterests::PipelineInterests(Face& face, const Options& options)
+PipelineInterests::PipelineInterests(Face& face)
: m_face(face)
- , m_nextSegmentNo(0)
, m_lastSegmentNo(0)
- , m_excludeSegmentNo(0)
- , m_options(options)
+ , m_excludedSegmentNo(0)
, m_hasFinalBlockId(false)
- , m_hasError(false)
- , m_hasFailure(false)
+ , m_isStopping(false)
{
- m_segmentFetchers.resize(m_options.maxPipelineSize);
}
-PipelineInterests::~PipelineInterests()
-{
- cancel();
-}
+PipelineInterests::~PipelineInterests() = default;
void
-PipelineInterests::runWithExcludedSegment(const Data& data, DataCallback onData,
- FailureCallback onFailure)
+PipelineInterests::run(const Data& data, DataCallback onData, FailureCallback onFailure)
{
BOOST_ASSERT(onData != nullptr);
m_onData = std::move(onData);
m_onFailure = std::move(onFailure);
-
- Name dataName = data.getName();
- m_prefix = dataName.getPrefix(-1);
- m_excludeSegmentNo = dataName[-1].toSegment();
+ m_prefix = data.getName().getPrefix(-1);
+ m_excludedSegmentNo = data.getName()[-1].toSegment();
if (!data.getFinalBlockId().empty()) {
- m_hasFinalBlockId = true;
m_lastSegmentNo = data.getFinalBlockId().toSegment();
+ m_hasFinalBlockId = true;
}
- // if the FinalBlockId is unknown, this could potentially request non-existent segments
- for (size_t nRequestedSegments = 0; nRequestedSegments < m_options.maxPipelineSize;
- nRequestedSegments++) {
- if (!fetchNextSegment(nRequestedSegments)) // all segments have been requested
- break;
- }
-}
-
-bool
-PipelineInterests::fetchNextSegment(std::size_t pipeNo)
-{
- if (m_hasFailure) {
- fail("Fetching terminated but no final segment number has been found");
- return false;
- }
-
- if (m_nextSegmentNo == m_excludeSegmentNo)
- m_nextSegmentNo++;
-
- if (m_hasFinalBlockId && m_nextSegmentNo > m_lastSegmentNo)
- return false;
-
- // Send interest for next segment
- if (m_options.isVerbose)
- std::cerr << "Requesting segment #" << m_nextSegmentNo << std::endl;
-
- Interest interest(Name(m_prefix).appendSegment(m_nextSegmentNo));
- interest.setInterestLifetime(m_options.interestLifetime);
- interest.setMustBeFresh(m_options.mustBeFresh);
- interest.setMaxSuffixComponents(1);
-
- BOOST_ASSERT(!m_segmentFetchers[pipeNo].first || !m_segmentFetchers[pipeNo].first->isRunning());
-
- auto fetcher = DataFetcher::fetch(m_face, interest,
- m_options.maxRetriesOnTimeoutOrNack,
- m_options.maxRetriesOnTimeoutOrNack,
- bind(&PipelineInterests::handleData, this, _1, _2, pipeNo),
- bind(&PipelineInterests::handleFail, this, _2, pipeNo),
- bind(&PipelineInterests::handleFail, this, _2, pipeNo),
- m_options.isVerbose);
-
- m_segmentFetchers[pipeNo] = make_pair(fetcher, m_nextSegmentNo);
-
- m_nextSegmentNo++;
- return true;
+ doRun();
}
void
PipelineInterests::cancel()
{
- for (auto& fetcher : m_segmentFetchers)
- if (fetcher.first)
- fetcher.first->cancel();
+ if (m_isStopping)
+ return;
- m_segmentFetchers.clear();
+ m_isStopping = true;
+ doCancel();
}
void
-PipelineInterests::fail(const std::string& reason)
+PipelineInterests::onFailure(const std::string& reason)
{
- if (!m_hasError) {
- cancel();
- m_hasError = true;
- m_hasFailure = true;
- if (m_onFailure)
- m_face.getIoService().post([this, reason] { m_onFailure(reason); });
- }
-}
-
-void
-PipelineInterests::handleData(const Interest& interest, const Data& data, size_t pipeNo)
-{
- if (m_hasError)
+ if (m_isStopping)
return;
- BOOST_ASSERT(data.getName().equals(interest.getName()));
+ cancel();
- if (m_options.isVerbose)
- std::cerr << "Received segment #" << data.getName()[-1].toSegment() << std::endl;
-
- m_onData(interest, data);
-
- if (!m_hasFinalBlockId && !data.getFinalBlockId().empty()) {
- m_lastSegmentNo = data.getFinalBlockId().toSegment();
- m_hasFinalBlockId = true;
-
- for (auto& fetcher : m_segmentFetchers) {
- if (fetcher.first && fetcher.second > m_lastSegmentNo) {
- // Stop trying to fetch segments that are not part of the content
- fetcher.first->cancel();
- }
- else if (fetcher.first && fetcher.first->hasError()) { // fetcher.second <= m_lastSegmentNo
- // there was an error while fetching a segment that is part of the content
- fail("Failure retriving segment #" + to_string(fetcher.second));
- return;
- }
- }
- }
-
- fetchNextSegment(pipeNo);
-}
-
-void PipelineInterests::handleFail(const std::string& reason, std::size_t pipeNo)
-{
- if (m_hasError)
- return;
-
- if (m_hasFinalBlockId && m_segmentFetchers[pipeNo].second <= m_lastSegmentNo) {
- fail(reason);
- }
- else if (!m_hasFinalBlockId) {
- // don't fetch the following segments
- bool areAllFetchersStopped = true;
- for (auto& fetcher : m_segmentFetchers) {
- if (fetcher.first && fetcher.second > m_segmentFetchers[pipeNo].second) {
- fetcher.first->cancel();
- }
- else if (fetcher.first && fetcher.first->isRunning()) {
- // fetcher.second <= m_segmentFetchers[pipeNo].second
- areAllFetchersStopped = false;
- }
- }
- if (areAllFetchersStopped) {
- if (m_onFailure)
- fail("Fetching terminated but no final segment number has been found");
- }
- else {
- m_hasFailure = true;
- }
- }
+ if (m_onFailure)
+ m_face.getIoService().post([this, reason] { m_onFailure(reason); });
}
} // namespace chunks