Semi-finalized Fetcher implementation, including unit test to check pipeline (assuming default pipeline of 6)
diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
index 2b43c62..cb94992 100644
--- a/src/fetch-manager.cc
+++ b/src/fetch-manager.cc
@@ -54,8 +54,20 @@
FetchManager::Enqueue (const Ccnx::Name &deviceName, uint32_t minSeqNo, uint32_t maxSeqNo, int priority/*=PRIORITY_NORMAL*/)
{
// we may need to guarantee that LookupLocator will gives an answer and not throw exception...
- Name forwardingHint = m_sync->LookupLocator (deviceName);
- Fetcher &fetcher = *(new Fetcher (*this, deviceName, minSeqNo, maxSeqNo, forwardingHint));
+ Name forwardingHint;
+ try {
+ forwardingHint = m_sync->LookupLocator (deviceName);
+ }
+ catch (Error::Db &exception) {
+ // just ignore for now
+ }
+
+ Fetcher &fetcher = *(new Fetcher (m_ccnx,
+ bind (&FetchManager::DidDataSegmentFetched, this, _1, _2, _3, _4, _5),
+ bind (&FetchManager::DidFetchComplete, this, _1),
+ bind (&FetchManager::DidNoDataTimeout, this, _1),
+ deviceName, minSeqNo, maxSeqNo
+ /* Alex: should or should not include hint initially?*/));
switch (priority)
{
@@ -81,7 +93,7 @@
m_currentParallelFetches < m_maxParallelFetches && item != m_fetchList.end ();
item++)
{
- if (item->m_active)
+ if (item->IsActive ())
continue;
m_currentParallelFetches ++;
@@ -90,10 +102,16 @@
}
void
+FetchManager::DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, const Bytes &data)
+{
+ // do something
+}
+
+void
FetchManager::DidNoDataTimeout (Fetcher &fetcher)
{
- fetcher.m_forwardingHint = Ccnx::Name ("/ndn/broadcast");
- fetcher.m_active = false;
+ fetcher.SetForwardingHint (Ccnx::Name ("/ndn/broadcast"));
{
unique_lock<mutex> lock (m_parellelFetchMutex);
m_currentParallelFetches --;
diff --git a/src/fetch-manager.h b/src/fetch-manager.h
index 73a0699..496d0fd 100644
--- a/src/fetch-manager.h
+++ b/src/fetch-manager.h
@@ -57,7 +57,11 @@
inline SchedulerPtr
GetScheduler ();
- // Events called from Fetcher
+ // Fetch Events
+ void
+ DidDataSegmentFetched (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, const Ccnx::Bytes &data);
+
void
DidNoDataTimeout (Fetcher &fetcher);
diff --git a/src/fetcher.cc b/src/fetcher.cc
index 6afad5b..876cab4 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -25,18 +25,29 @@
#include <boost/make_shared.hpp>
#include <boost/ref.hpp>
#include <boost/throw_exception.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost;
using namespace std;
using namespace Ccnx;
-Fetcher::Fetcher (FetchManager &fetchManger,
+Fetcher::Fetcher (CcnxWrapperPtr ccnx,
+ OnDataSegmentCallback onDataSegment,
+ OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+ boost::posix_time::time_duration timeout/* = boost::posix_time::seconds (30)*/,
const Ccnx::Name &forwardingHint/* = Ccnx::Name ()*/)
- : m_fetchManager (fetchManger)
+ : m_ccnx (ccnx)
+
+ , m_onDataSegment (onDataSegment)
+ , m_onFetchComplete (onFetchComplete)
+ , m_onFetchFailed (onFetchFailed)
+
, m_active (false)
, m_name (name)
, m_forwardingHint (forwardingHint)
+ , m_maximumNoActivityPeriod (timeout)
+
, m_minSendSeqNo (-1)
, m_maxInOrderRecvSeqNo (-1)
, m_minSeqNo (minSeqNo)
@@ -56,28 +67,37 @@
{
m_active = true;
m_minSendSeqNo = m_maxInOrderRecvSeqNo;
+ m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
FillPipeline ();
}
void
+Fetcher::SetForwardingHint (const Ccnx::Name &forwardingHint)
+{
+ m_forwardingHint = forwardingHint;
+}
+
+void
Fetcher::FillPipeline ()
{
for (; m_minSendSeqNo < m_maxSeqNo && m_activePipeline < m_pipeline; m_minSendSeqNo++)
{
- m_fetchManager.GetCcnx ()
- ->sendInterest (Name (m_name)("file")(m_minSendSeqNo+1),
- Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
- bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)));
+ m_ccnx->sendInterest (Name (m_name)(m_minSendSeqNo+1),
+ Closure (bind(&Fetcher::OnData, this, m_minSendSeqNo+1, _1, _2),
+ bind(&Fetcher::OnTimeout, this, m_minSendSeqNo+1, _1)));
m_activePipeline ++;
}
}
void
-Fetcher::OnData (uint32_t seqno, const Ccnx::Name &name, const Ccnx::Bytes &)
+Fetcher::OnData (uint32_t seqno, const Ccnx::Name &name, const Ccnx::Bytes &content)
{
+ m_onDataSegment (*this, seqno, m_name, name, content);
+
m_activePipeline --;
+ m_lastPositiveActivity = date_time::second_clock<boost::posix_time::ptime>::universal_time();
////////////////////////////////////////////////////////////////////////////
m_outOfOrderRecvSeqNo.insert (seqno);
@@ -95,18 +115,36 @@
m_outOfOrderRecvSeqNo.erase (m_outOfOrderRecvSeqNo.begin (), inOrderSeqNo);
////////////////////////////////////////////////////////////////////////////
- FillPipeline ();
- // bla bla
- if (0)
+ if (m_maxInOrderRecvSeqNo == m_maxSeqNo)
{
m_active = false;
- m_fetchManager.DidFetchComplete (*this);
+ m_onFetchComplete (*this);
+ }
+ else
+ {
+ FillPipeline ();
}
}
Closure::TimeoutCallbackReturnValue
Fetcher::OnTimeout (uint32_t seqno, const Ccnx::Name &name)
{
- // Closure::RESULT_REEXPRESS
- return Closure::RESULT_OK;
+ // cout << "Fetcher::OnTimeout: " << name << endl;
+ // cout << "Last: " << m_lastPositiveActivity << ", config: " << m_maximumNoActivityPeriod
+ // << ", now: " << date_time::second_clock<boost::posix_time::ptime>::universal_time()
+ // << ", oldest: " << (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod) << endl;
+
+ if (m_lastPositiveActivity <
+ (date_time::second_clock<boost::posix_time::ptime>::universal_time() - m_maximumNoActivityPeriod))
+ {
+ m_activePipeline --;
+ if (m_activePipeline == 0)
+ {
+ m_onFetchFailed (*this);
+ // this is not valid anymore, but we still should be able finish work
+ }
+ return Closure::RESULT_OK;
+ }
+ else
+ return Closure::RESULT_REEXPRESS;
}
diff --git a/src/fetcher.h b/src/fetcher.h
index c41c23c..271db52 100644
--- a/src/fetcher.h
+++ b/src/fetcher.h
@@ -27,22 +27,38 @@
#include "scheduler.h"
#include <boost/intrusive/list.hpp>
+#include <boost/date_time/posix_time/posix_time_types.hpp>
class FetchManager;
class Fetcher
{
public:
- Fetcher (FetchManager &fetchManger,
+ typedef boost::function<void (Fetcher &, uint32_t /*requested seqno*/, const Ccnx::Name & /*requested base name*/,
+ const Ccnx::Name & /*actual name*/, const Ccnx::Bytes &)> OnDataSegmentCallback;
+ typedef boost::function<void (Fetcher &)> OnFetchCompleteCallback;
+ typedef boost::function<void (Fetcher &)> OnFetchFailedCallback;
+
+ Fetcher (Ccnx::CcnxWrapperPtr ccnx,
+ OnDataSegmentCallback onDataSegment,
+ OnFetchCompleteCallback onFetchComplete, OnFetchFailedCallback onFetchFailed,
const Ccnx::Name &name, int32_t minSeqNo, int32_t maxSeqNo,
+ boost::posix_time::time_duration timeout = boost::posix_time::seconds (30), // this time is not precise, but sets min bound
+ // actual time depends on how fast Interests timeout
const Ccnx::Name &forwardingHint = Ccnx::Name ());
virtual ~Fetcher ();
-private:
+ inline bool
+ IsActive () const;
+
void
RestartPipeline ();
void
+ SetForwardingHint (const Ccnx::Name &forwardingHint);
+
+private:
+ void
FillPipeline ();
void
@@ -51,13 +67,23 @@
Ccnx::Closure::TimeoutCallbackReturnValue
OnTimeout (uint32_t seqno, const Ccnx::Name &name);
+public:
+ boost::intrusive::list_member_hook<> m_managerListHook;
+
private:
- FetchManager &m_fetchManager;
+ Ccnx::CcnxWrapperPtr m_ccnx;
+
+ OnDataSegmentCallback m_onDataSegment;
+ OnFetchCompleteCallback m_onFetchComplete;
+ OnFetchFailedCallback m_onFetchFailed;
+
bool m_active;
Ccnx::Name m_name;
Ccnx::Name m_forwardingHint;
+ boost::posix_time::time_duration m_maximumNoActivityPeriod;
+
int32_t m_minSendSeqNo;
int32_t m_maxInOrderRecvSeqNo;
std::set<int32_t> m_outOfOrderRecvSeqNo;
@@ -68,8 +94,8 @@
uint32_t m_pipeline;
uint32_t m_activePipeline;
- boost::intrusive::list_member_hook<> m_managerListHook;
- friend class FetchManager;
+ boost::posix_time::ptime m_lastPositiveActivity;
+
};
typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
@@ -80,5 +106,11 @@
typedef boost::shared_ptr<Fetcher> FetcherPtr;
+bool
+Fetcher::IsActive () const
+{
+ return m_active;
+}
+
#endif // FETCHER_H