Extension of scheduler, different logic in FetchManager regarding re-requesting (exponential backoff, and flipping broadcast/forwarding hint)
diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
index d74f210..e185581 100644
--- a/src/fetch-manager.cc
+++ b/src/fetch-manager.cc
@@ -23,6 +23,8 @@
#include <boost/make_shared.hpp>
#include <boost/ref.hpp>
#include <boost/throw_exception.hpp>
+
+#include "simple-interval-generator.h"
#include "logging.h"
INIT_LOGGER ("FetchManager");
@@ -31,21 +33,30 @@
using namespace std;
using namespace Ccnx;
-static const string BROADCAST_DOMAIN = "/ndn/broadcast/chronoshare";
+static const Name BROADCAST_DOMAIN = Name ("/ndn/broadcast/chronoshare");
//The disposer object function
struct fetcher_disposer { void operator() (Fetcher *delete_this) { delete delete_this; } };
+static const string SCHEDULE_FETCHES_TAG = "ScheduleFetches";
+
FetchManager::FetchManager (CcnxWrapperPtr ccnx, const Mapping &mapping, uint32_t parallelFetches/* = 3*/)
: m_ccnx (ccnx)
, m_mapping (mapping)
, m_maxParallelFetches (parallelFetches)
, m_currentParallelFetches (0)
-
+ , m_scheduler (new Scheduler)
{
+ m_scheduler->start ();
+
+ m_scheduleFetchesTask = Scheduler::schedulePeriodicTask (m_scheduler,
+ make_shared<SimpleIntervalGenerator> (1),
+ bind (&FetchManager::ScheduleFetches, this), SCHEDULE_FETCHES_TAG);
}
FetchManager::~FetchManager ()
{
+ m_scheduler->shutdown ();
+
m_fetchList.clear_and_dispose (fetcher_disposer ());
}
@@ -85,7 +96,8 @@
break;
}
- ScheduleFetches (); // will start a fetch if m_currentParallelFetches is less than max, otherwise does nothing
+ m_scheduler->rescheduleTaskAt (m_scheduleFetchesTask, 0);
+ // ScheduleFetches (); // will start a fetch if m_currentParallelFetches is less than max, otherwise does nothing
}
void
@@ -93,6 +105,8 @@
{
unique_lock<mutex> lock (m_parellelFetchMutex);
+ boost::posix_time::ptime currentTime = date_time::second_clock<boost::posix_time::ptime>::universal_time ();
+
for (FetchList::iterator item = m_fetchList.begin ();
m_currentParallelFetches < m_maxParallelFetches && item != m_fetchList.end ();
item++)
@@ -100,6 +114,9 @@
if (item->IsActive ())
continue;
+ if (currentTime < item->GetNextScheduledRetry ())
+ continue;
+
_LOG_DEBUG ("Start fetching of " << item->GetName ());
m_currentParallelFetches ++;
@@ -112,14 +129,38 @@
{
_LOG_DEBUG ("No data timeout for " << fetcher.GetName () << " with forwarding hint: " << fetcher.GetForwardingHint ());
- fetcher.SetForwardingHint (Ccnx::Name (BROADCAST_DOMAIN));
{
unique_lock<mutex> lock (m_parellelFetchMutex);
m_currentParallelFetches --;
// no need to do anything with the m_fetchList
}
- ScheduleFetches ();
+ if (fetcher.GetForwardingHint () == BROADCAST_DOMAIN)
+ {
+ // try again directly (hopefully with different forwarding hint
+
+ /// @todo Handle potential exception
+ Name forwardingHint;
+ forwardingHint = m_mapping (fetcher.GetDeviceName ());
+ fetcher.SetForwardingHint (forwardingHint);
+ }
+ else
+ {
+ fetcher.SetForwardingHint (BROADCAST_DOMAIN);
+ }
+
+ double delay = fetcher.GetRetryPause ();
+ if (delay < 1) // first time
+ {
+ delay = 1;
+ }
+ else
+ {
+ delay = std::min (2*delay, 300.0); // 5 minutes max
+ }
+
+ fetcher.SetRetryPause (delay);
+ fetcher.SetNextScheduledRetry (date_time::second_clock<boost::posix_time::ptime>::universal_time () + posix_time::seconds (delay));
}
void
diff --git a/src/fetch-manager.h b/src/fetch-manager.h
index c26b78d..329b289 100644
--- a/src/fetch-manager.h
+++ b/src/fetch-manager.h
@@ -58,10 +58,6 @@
GetCcnx ();
private:
-
- inline SchedulerPtr
- GetScheduler ();
-
// Fetch Events
void
DidDataSegmentFetched (Fetcher &fetcher, uint64_t seqno, const Ccnx::Name &basename,
@@ -90,6 +86,8 @@
typedef boost::intrusive::list<Fetcher, MemberOption> FetchList;
FetchList m_fetchList;
+ SchedulerPtr m_scheduler;
+ TaskPtr m_scheduleFetchesTask;
};
Ccnx::CcnxWrapperPtr
diff --git a/src/fetcher.cc b/src/fetcher.cc
index 3fe7c63..9c1475d 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -62,15 +62,13 @@
, m_pipeline (6) // initial "congestion window"
, m_activePipeline (0)
-
- , m_executor (1)
+ , m_retryPause (0)
+ , m_nextScheduledRetry () // zero time
{
- m_executor.start ();
}
Fetcher::~Fetcher ()
{
- m_executor.shutdown ();
}
void
@@ -81,7 +79,10 @@
// cout << "Restart: " << m_minSendSeqNo << endl;
m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
- m_executor.execute (bind (&Fetcher::FillPipeline, this));
+ // Scheduler::scheduleOneTimeTask ();
+ // m_scheduler
+ // m_executor.execute (bind (&Fetcher::FillPipeline, this));
+ FillPipeline ();
}
void
@@ -175,7 +176,8 @@
}
else
{
- m_executor.execute (bind (&Fetcher::FillPipeline, this));
+ FillPipeline ();
+ // m_executor.execute (bind (&Fetcher::FillPipeline, this));
}
}
@@ -183,7 +185,7 @@
Fetcher::OnTimeout (uint64_t seqno, const Ccnx::Name &name)
{
_LOG_DEBUG (" <<< :( timeout " << name.getPartialName (0, name.size () - 1) << ", seq = " << seqno);
-
+
// cout << "Fetcher::OnTimeout: " << name << endl;
// cout << "Last: " << m_lastPositiveActivity << ", config: " << m_maximumNoActivityPeriod
// << ", now: " << date_time::second_clock<boost::posix_time::ptime>::universal_time()
diff --git a/src/fetcher.h b/src/fetcher.h
index 1be6a27..6a31b5e 100644
--- a/src/fetcher.h
+++ b/src/fetcher.h
@@ -64,6 +64,20 @@
const Ccnx::Name &
GetName () const { return m_name; }
+ const Ccnx::Name &
+ GetDeviceName () const { return m_deviceName; }
+
+ double
+ GetRetryPause () const { return m_retryPause; }
+
+ void
+ SetRetryPause (double pause) { m_retryPause = pause; }
+
+ boost::posix_time::ptime
+ GetNextScheduledRetry () const { return m_nextScheduledRetry; }
+
+ void
+ SetNextScheduledRetry (boost::posix_time::ptime nextScheduledRetry) { m_nextScheduledRetry = nextScheduledRetry; }
private:
void
@@ -106,7 +120,9 @@
uint32_t m_activePipeline;
boost::posix_time::ptime m_lastPositiveActivity;
- Executor m_executor;
+
+ double m_retryPause; // pause to stop trying to fetch (for fetch-manager)
+ boost::posix_time::ptime m_nextScheduledRetry;
};
typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;