Semi-finalized Fetcher implementation, including unit test to check pipeline (assuming default pipeline of 6)
diff --git a/ccnx/ccnx-wrapper.h b/ccnx/ccnx-wrapper.h
index 2e66a2d..1127316 100644
--- a/ccnx/ccnx-wrapper.h
+++ b/ccnx/ccnx-wrapper.h
@@ -41,10 +41,10 @@
sendInterest (const Name &interest, const Closure &closure, const Selectors &selector = Selectors());
virtual int
- publishData (const Name &name, const unsigned char *buf, size_t len, int freshness);
+ publishData (const Name &name, const unsigned char *buf, size_t len, int freshness = 2147/* max value for ccnx*/);
int
- publishData (const Name &name, const Bytes &content, int freshness);
+ publishData (const Name &name, const Bytes &content, int freshness = 2147/* max value for ccnx*/);
static Name
getLocalPrefix ();
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
diff --git a/test/test-fetch-manager.cc b/test/test-fetch-manager.cc
new file mode 100644
index 0000000..d28570e
--- /dev/null
+++ b/test/test-fetch-manager.cc
@@ -0,0 +1,162 @@
+/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
+/*
+ * Copyright (c) 2012 University of California, Los Angeles
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation;
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ *
+ * Author: Alexander Afanasyev <alexander.afanasyev@ucla.edu>
+ * Zhenkai Zhu <zhenkai@cs.ucla.edu>
+ */
+
+#include "fetch-manager.h"
+#include "fetcher.h"
+#include "ccnx-wrapper.h"
+#include <boost/test/unit_test.hpp>
+#include <boost/make_shared.hpp>
+
+using namespace Ccnx;
+using namespace std;
+using namespace boost;
+
+BOOST_AUTO_TEST_SUITE(TestFetchManager)
+
+struct FetcherTestData
+{
+ set<uint32_t> recvData;
+ set<uint32_t> recvContent;
+
+ set<Name> differentNames;
+ set<Name> segmentNames;
+
+ bool m_done;
+ bool m_failed;
+
+ FetcherTestData ()
+ : m_done (false)
+ , m_failed (false)
+ {
+ }
+
+ void
+ onData (Fetcher &fetcher, uint32_t seqno, const Ccnx::Name &basename,
+ const Ccnx::Name &name, const Ccnx::Bytes &data)
+ {
+ recvData.insert (seqno);
+ differentNames.insert (basename);
+ segmentNames.insert (name);
+
+ if (data.size () == sizeof(int))
+ {
+ recvContent.insert (*reinterpret_cast<const int*> (head(data)));
+ }
+
+ // cout << basename << ", " << name << ", " << seqno << endl;
+ }
+
+ void
+ onComplete (Fetcher &fetcher)
+ {
+ m_done = true;
+ // cout << "Done" << endl;
+ }
+
+ void
+ onFail (Fetcher &fetcher)
+ {
+ m_failed = true;
+ // cout << "Failed" << endl;
+ }
+};
+
+
+BOOST_AUTO_TEST_CASE (TestFetcher)
+{
+ CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
+
+ Name baseName ("/base");
+ /* publish seqnos: 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, <gap 5>, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, <gap 1>, 26 */
+ // this will allow us to test our pipeline of 6
+ for (int i = 0; i < 10; i++)
+ {
+ ccnx->publishData (Name (baseName)(i), reinterpret_cast<const unsigned char*> (&i), sizeof(int), 10);
+
+ int other = 10 + i+5;
+ ccnx->publishData (Name (baseName)(other), reinterpret_cast<const unsigned char*> (&other), sizeof(int), 10);
+ }
+
+ int oneMore = 26;
+ ccnx->publishData (Name (baseName)(oneMore), reinterpret_cast<const unsigned char*> (&oneMore), sizeof(int), 10);
+
+ FetcherTestData data;
+
+ Fetcher fetcher (ccnx,
+ bind (&FetcherTestData::onData, &data, _1, _2, _3, _4, _5),
+ bind (&FetcherTestData::onComplete, &data, _1),
+ bind (&FetcherTestData::onFail, &data, _1),
+ Name ("/base"), 0, 26,
+ boost::posix_time::seconds (5)); // this time is not precise
+
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), false);
+ fetcher.RestartPipeline ();
+ BOOST_CHECK_EQUAL (fetcher.IsActive (), true);
+
+ usleep(13000000);
+ BOOST_CHECK_EQUAL (data.m_failed, true);
+ BOOST_CHECK_EQUAL (data.differentNames.size (), 1);
+ BOOST_CHECK_EQUAL (data.segmentNames.size (), 10);
+ BOOST_CHECK_EQUAL (data.recvData.size (), 10);
+ BOOST_CHECK_EQUAL (data.recvContent.size (), 10);
+
+ ostringstream recvData;
+ for (set<uint32_t>::iterator i = data.recvData.begin (); i != data.recvData.end (); i++)
+ recvData << *i << ", ";
+
+ ostringstream recvContent;
+ for (set<uint32_t>::iterator i = data.recvContent.begin (); i != data.recvContent.end (); i++)
+ recvContent << *i << ", ";
+
+ BOOST_CHECK_EQUAL (recvData.str (), recvContent.str ());
+}
+
+// BOOST_AUTO_TEST_CASE (CcnxWrapperSelector)
+// {
+
+// Closure closure (bind(dataCallback, _1, _2), bind(timeout, _1));
+
+// Selectors selectors;
+// selectors.interestLifetime(1);
+
+// string n1 = "/random/01";
+// c1->sendInterest(Name(n1), closure, selectors);
+// sleep(2);
+// c2->publishData(Name(n1), (const unsigned char *)n1.c_str(), n1.size(), 4);
+// usleep(100000);
+// BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+// BOOST_CHECK_EQUAL(g_dataCallback_counter, 0);
+
+// string n2 = "/random/02";
+// selectors.interestLifetime(2);
+// c1->sendInterest(Name(n2), closure, selectors);
+// sleep(1);
+// c2->publishData(Name(n2), (const unsigned char *)n2.c_str(), n2.size(), 4);
+// usleep(100000);
+// BOOST_CHECK_EQUAL(g_timeout_counter, 1);
+// BOOST_CHECK_EQUAL(g_dataCallback_counter, 1);
+
+// // reset
+// g_dataCallback_counter = 0;
+// g_timeout_counter = 0;
+// }
+
+BOOST_AUTO_TEST_SUITE_END()