face: Refactor internal PIT to use scheduled events
The commit also changes how face is paused when there are no pending
interests left and there are no registered prefixes with local
forwarder: data structures for pending interests and registered prefixes
will fire up a signal when they become empty.
Change-Id: I6b87a44b0c8bc766865a51962ecacaec85b4adad
Refs: #1372, #2518
diff --git a/src/detail/container-with-on-empty-signal.hpp b/src/detail/container-with-on-empty-signal.hpp
new file mode 100644
index 0000000..0c3e7c3
--- /dev/null
+++ b/src/detail/container-with-on-empty-signal.hpp
@@ -0,0 +1,108 @@
+/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
+/**
+ * Copyright (c) 2013-2015 Regents of the University of California.
+ *
+ * This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
+ *
+ * ndn-cxx library is free software: you can redistribute it and/or modify it under the
+ * terms of the GNU Lesser General Public License as published by the Free Software
+ * Foundation, either version 3 of the License, or (at your option) any later version.
+ *
+ * ndn-cxx library 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 Lesser General Public License for more details.
+ *
+ * You should have received copies of the GNU General Public License and GNU Lesser
+ * General Public License along with ndn-cxx, e.g., in COPYING.md file. If not, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
+ */
+
+#ifndef NDN_DETAIL_CONTAINER_WITH_ON_EMPTY_SIGNAL_HPP
+#define NDN_DETAIL_CONTAINER_WITH_ON_EMPTY_SIGNAL_HPP
+
+#include "../common.hpp"
+#include "../util/signal.hpp"
+
+namespace ndn {
+
+/**
+ * @brief A simple container that will fire up onEmpty signal when there are no entries left
+ */
+template<class T>
+class ContainerWithOnEmptySignal
+{
+public:
+ typedef std::list<T> Base;
+ typedef typename Base::value_type value_type;
+ typedef typename Base::iterator iterator;
+
+ iterator
+ begin()
+ {
+ return m_container.begin();
+ }
+
+ iterator
+ end()
+ {
+ return m_container.end();
+ }
+
+ size_t
+ size()
+ {
+ return m_container.size();
+ }
+
+ bool
+ empty()
+ {
+ return m_container.empty();
+ }
+
+ iterator
+ erase(iterator item)
+ {
+ iterator next = m_container.erase(item);
+ if (empty()) {
+ this->onEmpty();
+ }
+ return next;
+ }
+
+ void
+ clear()
+ {
+ m_container.clear();
+ this->onEmpty();
+ }
+
+ std::pair<iterator, bool>
+ insert(const value_type& value)
+ {
+ return {m_container.insert(end(), value), true};
+ }
+
+ template<class Predicate>
+ void remove_if(Predicate p)
+ {
+ m_container.remove_if(p);
+ if (empty()) {
+ this->onEmpty();
+ }
+ }
+
+public:
+ Base m_container;
+
+ /**
+ * @brief Signal to be fired when container becomes empty
+ */
+ util::Signal<ContainerWithOnEmptySignal<T>> onEmpty;
+};
+
+} // namespace ndn
+
+#endif // NDN_DETAIL_CONTAINER_WITH_ON_EMPTY_SIGNAL_HPP
diff --git a/src/detail/face-impl.hpp b/src/detail/face-impl.hpp
index cb5b3bc..fb6642a 100644
--- a/src/detail/face-impl.hpp
+++ b/src/detail/face-impl.hpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2014 Regents of the University of California.
+ * Copyright (c) 2013-2015 Regents of the University of California.
*
* This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
*
@@ -27,9 +27,11 @@
#include "registered-prefix.hpp"
#include "pending-interest.hpp"
+#include "container-with-on-empty-signal.hpp"
#include "../util/scheduler.hpp"
#include "../util/config-file.hpp"
+#include "../util/signal.hpp"
#include "../transport/transport.hpp"
#include "../transport/unix-transport.hpp"
@@ -43,14 +45,26 @@
class Face::Impl : noncopyable
{
public:
- typedef std::list<shared_ptr<PendingInterest> > PendingInterestTable;
+ typedef ContainerWithOnEmptySignal<shared_ptr<PendingInterest>> PendingInterestTable;
typedef std::list<shared_ptr<InterestFilterRecord> > InterestFilterTable;
- typedef std::list<shared_ptr<RegisteredPrefix> > RegisteredPrefixTable;
+ typedef ContainerWithOnEmptySignal<shared_ptr<RegisteredPrefix>> RegisteredPrefixTable;
explicit
Impl(Face& face)
: m_face(face)
+ , m_scheduler(m_face.getIoService())
+ , m_processEventsTimeoutEvent(m_scheduler)
{
+ auto postOnEmptyPitOrNoRegisteredPrefixes = [this] {
+ this->m_face.getIoService().post(bind(&Impl::onEmptyPitOrNoRegisteredPrefixes, this));
+ // without this extra "post", transport can get paused (-async_read) and then resumed
+ // (+async_read) from within onInterest/onData callback. After onInterest/onData
+ // finishes, there is another +async_read with the same memory block. A few of such
+ // async_read duplications can cause various effects and result in segfault.
+ };
+
+ m_pendingInterestTable.onEmpty.connect(postOnEmptyPitOrNoRegisteredPrefixes);
+ m_registeredPrefixTable.onEmpty.connect(postOnEmptyPitOrNoRegisteredPrefixes);
}
/////////////////////////////////////////////////////////////////////////////////////////////////
@@ -59,49 +73,34 @@
void
satisfyPendingInterests(Data& data)
{
- for (PendingInterestTable::iterator i = m_pendingInterestTable.begin();
- i != m_pendingInterestTable.end();
- )
- {
- if ((*i)->getInterest()->matchesData(data))
- {
- // Copy pointers to the objects and remove the PIT entry before calling the callback.
- OnData onData = (*i)->getOnData();
- shared_ptr<const Interest> interest = (*i)->getInterest();
+ for (auto entry = m_pendingInterestTable.begin(); entry != m_pendingInterestTable.end(); ) {
+ if ((*entry)->getInterest().matchesData(data)) {
+ shared_ptr<PendingInterest> matchedEntry = *entry;
- PendingInterestTable::iterator next = i;
- ++next;
- m_pendingInterestTable.erase(i);
- i = next;
+ entry = m_pendingInterestTable.erase(entry);
- if (static_cast<bool>(onData)) {
- onData(*interest, data);
- }
- }
- else
- ++i;
+ matchedEntry->invokeDataCallback(data);
}
+ else
+ ++entry;
+ }
}
void
processInterestFilters(Interest& interest)
{
- for (InterestFilterTable::iterator i = m_interestFilterTable.begin();
- i != m_interestFilterTable.end();
- ++i)
- {
- if ((*i)->doesMatch(interest.getName()))
- {
- (**i)(interest);
- }
+ for (const auto& filter : m_interestFilterTable) {
+ if (filter->doesMatch(interest.getName())) {
+ filter->invokeInterestCallback(interest);
}
+ }
}
/////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////
void
- ensureConnected(bool wantResume = true)
+ ensureConnected(bool wantResume)
{
if (!m_face.m_transport->isConnected())
m_face.m_transport->connect(m_face.m_ioService,
@@ -115,26 +114,22 @@
asyncExpressInterest(const shared_ptr<const Interest>& interest,
const OnData& onData, const OnTimeout& onTimeout)
{
- this->ensureConnected();
+ this->ensureConnected(true);
- m_pendingInterestTable.push_back(make_shared<PendingInterest>(interest, onData, onTimeout));
+ auto entry =
+ m_pendingInterestTable.insert(make_shared<PendingInterest>(interest,
+ onData, onTimeout,
+ ref(m_scheduler))).first;
+ (*entry)->setDeleter([this, entry] { m_pendingInterestTable.erase(entry); });
- if (!interest->getLocalControlHeader().empty(nfd::LocalControlHeader::ENCODE_NEXT_HOP))
- {
- // encode only NextHopFaceId towards the forwarder
- m_face.m_transport->send(interest->getLocalControlHeader()
- .wireEncode(*interest, nfd::LocalControlHeader::ENCODE_NEXT_HOP),
- interest->wireEncode());
- }
- else
- {
- m_face.m_transport->send(interest->wireEncode());
- }
-
- if (!m_pitTimeoutCheckTimerActive) {
- m_pitTimeoutCheckTimerActive = true;
- m_pitTimeoutCheckTimer->expires_from_now(time::milliseconds(100));
- m_pitTimeoutCheckTimer->async_wait(bind(&Impl::checkPitExpire, this));
+ if (!interest->getLocalControlHeader().empty(nfd::LocalControlHeader::ENCODE_NEXT_HOP)) {
+ // encode only NextHopFaceId towards the forwarder
+ m_face.m_transport->send(interest->getLocalControlHeader()
+ .wireEncode(*interest, nfd::LocalControlHeader::ENCODE_NEXT_HOP),
+ interest->wireEncode());
+ }
+ else {
+ m_face.m_transport->send(interest->wireEncode());
}
}
@@ -147,19 +142,17 @@
void
asyncPutData(const shared_ptr<const Data>& data)
{
- this->ensureConnected();
+ this->ensureConnected(true);
- if (!data->getLocalControlHeader().empty(nfd::LocalControlHeader::ENCODE_CACHING_POLICY))
- {
- m_face.m_transport->send(
- data->getLocalControlHeader().wireEncode(*data,
- nfd::LocalControlHeader::ENCODE_CACHING_POLICY),
- data->wireEncode());
- }
- else
- {
- m_face.m_transport->send(data->wireEncode());
- }
+ if (!data->getLocalControlHeader().empty(nfd::LocalControlHeader::ENCODE_CACHING_POLICY)) {
+ m_face.m_transport->send(
+ data->getLocalControlHeader().wireEncode(*data,
+ nfd::LocalControlHeader::ENCODE_CACHING_POLICY),
+ data->wireEncode());
+ }
+ else {
+ m_face.m_transport->send(data->wireEncode());
+ }
}
/////////////////////////////////////////////////////////////////////////////////////////////////
@@ -215,7 +208,7 @@
afterPrefixRegistered(const shared_ptr<RegisteredPrefix>& registeredPrefix,
const RegisterPrefixSuccessCallback& onSuccess)
{
- m_registeredPrefixTable.push_back(registeredPrefix);
+ m_registeredPrefixTable.insert(registeredPrefix);
if (static_cast<bool>(registeredPrefix->getFilter())) {
// it was a combined operation
@@ -268,75 +261,32 @@
{
m_registeredPrefixTable.erase(item);
- if (!m_pitTimeoutCheckTimerActive && m_registeredPrefixTable.empty())
- {
- m_face.m_transport->pause();
- if (!m_ioServiceWork) {
- m_processEventsTimeoutTimer->cancel();
- }
- }
-
if (static_cast<bool>(onSuccess)) {
onSuccess();
}
}
- /////////////////////////////////////////////////////////////////////////////////////////////////
- /////////////////////////////////////////////////////////////////////////////////////////////////
-
void
- checkPitExpire()
+ onEmptyPitOrNoRegisteredPrefixes()
{
- // Check for PIT entry timeouts.
- time::steady_clock::TimePoint now = time::steady_clock::now();
-
- PendingInterestTable::iterator i = m_pendingInterestTable.begin();
- while (i != m_pendingInterestTable.end())
- {
- if ((*i)->isTimedOut(now))
- {
- // Save the PendingInterest and remove it from the PIT. Then call the callback.
- shared_ptr<PendingInterest> pendingInterest = *i;
-
- i = m_pendingInterestTable.erase(i);
-
- pendingInterest->callTimeout();
- }
- else
- ++i;
- }
-
- if (!m_pendingInterestTable.empty()) {
- m_pitTimeoutCheckTimerActive = true;
-
- m_pitTimeoutCheckTimer->expires_from_now(time::milliseconds(100));
- m_pitTimeoutCheckTimer->async_wait(bind(&Impl::checkPitExpire, this));
- }
- else {
- m_pitTimeoutCheckTimerActive = false;
-
- if (m_registeredPrefixTable.empty()) {
- m_face.m_transport->pause();
- if (!m_ioServiceWork) {
- m_processEventsTimeoutTimer->cancel();
- }
+ if (m_pendingInterestTable.empty() && m_registeredPrefixTable.empty()) {
+ m_face.m_transport->pause();
+ if (!m_ioServiceWork) {
+ m_processEventsTimeoutEvent.cancel();
}
}
}
private:
Face& m_face;
+ util::Scheduler m_scheduler;
+ util::scheduler::ScopedEventId m_processEventsTimeoutEvent;
PendingInterestTable m_pendingInterestTable;
InterestFilterTable m_interestFilterTable;
RegisteredPrefixTable m_registeredPrefixTable;
- ConfigFile m_config;
-
- shared_ptr<boost::asio::io_service::work> m_ioServiceWork; // if thread needs to be preserved
- shared_ptr<monotonic_deadline_timer> m_pitTimeoutCheckTimer;
- bool m_pitTimeoutCheckTimerActive;
- shared_ptr<monotonic_deadline_timer> m_processEventsTimeoutTimer;
+ unique_ptr<boost::asio::io_service::work> m_ioServiceWork; // if thread needs to be preserved
friend class Face;
};
diff --git a/src/detail/interest-filter-record.hpp b/src/detail/interest-filter-record.hpp
index 2f8d118..f3a224c 100644
--- a/src/detail/interest-filter-record.hpp
+++ b/src/detail/interest-filter-record.hpp
@@ -31,11 +31,11 @@
class InterestFilterRecord : noncopyable
{
public:
- typedef function<void (const InterestFilter&, const Interest&)> OnInterest;
+ typedef function<void (const InterestFilter&, const Interest&)> InterestCallback;
- InterestFilterRecord(const InterestFilter& filter, const OnInterest& onInterest)
+ InterestFilterRecord(const InterestFilter& filter, const InterestCallback& afterInterest)
: m_filter(filter)
- , m_onInterest(onInterest)
+ , m_afterInterest(afterInterest)
{
}
@@ -49,10 +49,14 @@
return m_filter.doesMatch(name);
}
+ /**
+ * @brief invokes the InterestCallback
+ * @note If the DataCallback is an empty function, this method does nothing.
+ */
void
- operator()(const Interest& interest) const
+ invokeInterestCallback(const Interest& interest) const
{
- m_onInterest(m_filter, interest);
+ m_afterInterest(m_filter, interest);
}
const InterestFilter&
@@ -63,7 +67,7 @@
private:
InterestFilter m_filter;
- OnInterest m_onInterest;
+ InterestCallback m_afterInterest;
};
diff --git a/src/detail/pending-interest.hpp b/src/detail/pending-interest.hpp
index 0a66dca..41f6b37 100644
--- a/src/detail/pending-interest.hpp
+++ b/src/detail/pending-interest.hpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2014 Regents of the University of California.
+ * Copyright (c) 2013-2015 Regents of the University of California.
*
* This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
*
@@ -26,6 +26,8 @@
#include "../interest.hpp"
#include "../data.hpp"
#include "../util/time.hpp"
+#include "../util/scheduler.hpp"
+#include "../util/scheduler-scoped-event-id.hpp"
namespace ndn {
@@ -43,57 +45,73 @@
* @param onData A function object to call when a matching data packet is received.
* @param onTimeout A function object to call if the interest times out.
* If onTimeout is an empty OnTimeout(), this does not use it.
+ * @param scheduler Scheduler instance to use to schedule a timeout event. The scheduled
+ * event will be automatically cancelled when pending interest is destroyed.
*/
- PendingInterest(const shared_ptr<const Interest>& interest, const OnData& onData,
- const OnTimeout& onTimeout)
+ PendingInterest(shared_ptr<const Interest> interest, const OnData& onData,
+ const OnTimeout& onTimeout, Scheduler& scheduler)
: m_interest(interest)
, m_onData(onData)
, m_onTimeout(onTimeout)
+ , m_timeoutEvent(scheduler)
{
- if (m_interest->getInterestLifetime() >= time::milliseconds::zero())
- m_timeout = time::steady_clock::now() + m_interest->getInterestLifetime();
- else
- m_timeout = time::steady_clock::now() + DEFAULT_INTEREST_LIFETIME;
+ m_timeoutEvent =
+ scheduler.scheduleEvent(m_interest->getInterestLifetime() > time::milliseconds::zero() ?
+ m_interest->getInterestLifetime() :
+ DEFAULT_INTEREST_LIFETIME,
+ bind(&PendingInterest::invokeTimeoutCallback, this));
}
- const shared_ptr<const Interest>&
+ /**
+ * @return the Interest
+ */
+ const Interest&
getInterest() const
{
- return m_interest;
- }
-
- const OnData&
- getOnData() const
- {
- return m_onData;
+ return *m_interest;
}
/**
- * Check if this interest is timed out.
- * @return true if this interest timed out, otherwise false.
- */
- bool
- isTimedOut(const time::steady_clock::TimePoint& now) const
- {
- return now >= m_timeout;
- }
-
- /**
- * Call m_onTimeout (if defined). This ignores exceptions from the m_onTimeout.
+ * @brief invokes the DataCallback
+ * @note If the DataCallback is an empty function, this method does nothing.
*/
void
- callTimeout() const
+ invokeDataCallback(Data& data)
+ {
+ m_onData(*m_interest, data);
+ }
+
+ /**
+ * @brief Set cleanup function to be called after interest times out
+ */
+ void
+ setDeleter(const std::function<void()>& deleter)
+ {
+ m_deleter = deleter;
+ }
+
+private:
+ /**
+ * @brief invokes the TimeoutCallback
+ * @note If the TimeoutCallback is an empty function, this method does nothing.
+ */
+ void
+ invokeTimeoutCallback()
{
if (m_onTimeout) {
m_onTimeout(*m_interest);
}
+
+ BOOST_ASSERT(m_deleter);
+ m_deleter();
}
private:
shared_ptr<const Interest> m_interest;
const OnData m_onData;
const OnTimeout m_onTimeout;
- time::steady_clock::TimePoint m_timeout;
+ util::scheduler::ScopedEventId m_timeoutEvent;
+ std::function<void()> m_deleter;
};
@@ -115,7 +133,7 @@
operator()(const shared_ptr<const PendingInterest>& pendingInterest) const
{
return (reinterpret_cast<const PendingInterestId*>(
- pendingInterest->getInterest().get()) == m_id);
+ &pendingInterest->getInterest()) == m_id);
}
private:
const PendingInterestId* m_id;
diff --git a/src/face.cpp b/src/face.cpp
index b419e13..be2aaa1 100644
--- a/src/face.cpp
+++ b/src/face.cpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2014 Regents of the University of California.
+ * Copyright (c) 2013-2015 Regents of the University of California.
*
* This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
*
@@ -17,8 +17,6 @@
* <http://www.gnu.org/licenses/>.
*
* See AUTHORS.md for complete list of ndn-cxx authors and contributors.
- *
- * Based on code originally written by Jeff Thompson <jefft0@remap.ucla.edu>
*/
#include "face.hpp"
@@ -93,41 +91,34 @@
// transport=unix:///var/run/nfd.sock
// transport=tcp://localhost:6363
- const ConfigFile::Parsed& parsed = m_impl->m_config.getParsedConfiguration();
-
- const auto transportType = parsed.get_optional<std::string>("transport");
- if (!transportType)
- {
- // transport not specified, use default Unix transport.
- construct(UnixTransport::create(m_impl->m_config), keyChain);
- return;
- }
+ ConfigFile config;
+ const auto& transportType = config.getParsedConfiguration()
+ .get_optional<std::string>("transport");
+ if (!transportType) {
+ // transport not specified, use default Unix transport.
+ construct(UnixTransport::create(config), keyChain);
+ return;
+ }
unique_ptr<util::FaceUri> uri;
- try
- {
- uri.reset(new util::FaceUri(*transportType));
- }
- catch (const util::FaceUri::Error& error)
- {
- throw ConfigFile::Error(error.what());
- }
+ try {
+ uri.reset(new util::FaceUri(*transportType));
+ }
+ catch (const util::FaceUri::Error& error) {
+ throw ConfigFile::Error(error.what());
+ }
const std::string protocol = uri->getScheme();
- if (protocol == "unix")
- {
- construct(UnixTransport::create(m_impl->m_config), keyChain);
-
- }
- else if (protocol == "tcp" || protocol == "tcp4" || protocol == "tcp6")
- {
- construct(TcpTransport::create(m_impl->m_config), keyChain);
- }
- else
- {
- throw ConfigFile::Error("Unsupported transport protocol \"" + protocol + "\"");
- }
+ if (protocol == "unix") {
+ construct(UnixTransport::create(config), keyChain);
+ }
+ else if (protocol == "tcp" || protocol == "tcp4" || protocol == "tcp6") {
+ construct(TcpTransport::create(config), keyChain);
+ }
+ else {
+ throw ConfigFile::Error("Unsupported transport protocol \"" + protocol + "\"");
+ }
}
void
@@ -135,11 +126,8 @@
{
m_nfdController.reset(new nfd::Controller(*this, keyChain));
- m_impl->m_pitTimeoutCheckTimerActive = false;
m_transport = transport;
- m_impl->m_pitTimeoutCheckTimer = make_shared<monotonic_deadline_timer>(ref(m_ioService));
- m_impl->m_processEventsTimeoutTimer = make_shared<monotonic_deadline_timer>(ref(m_ioService));
m_impl->ensureConnected(false);
}
@@ -358,30 +346,29 @@
}
try {
- if (timeout < time::milliseconds::zero())
- {
+ if (timeout < time::milliseconds::zero()) {
// do not block if timeout is negative, but process pending events
m_ioService.poll();
return;
}
- if (timeout > time::milliseconds::zero())
- {
- m_impl->m_processEventsTimeoutTimer->expires_from_now(time::milliseconds(timeout));
- m_impl->m_processEventsTimeoutTimer->async_wait(&fireProcessEventsTimeout);
- }
+ if (timeout > time::milliseconds::zero()) {
+ boost::asio::io_service& ioService = m_ioService;
+ unique_ptr<boost::asio::io_service::work>& work = m_impl->m_ioServiceWork;
+ m_impl->m_processEventsTimeoutEvent =
+ m_impl->m_scheduler.scheduleEvent(timeout, [&ioService, &work] {
+ ioService.stop();
+ work.reset();
+ });
+ }
if (keepThread) {
// work will ensure that m_ioService is running until work object exists
- m_impl->m_ioServiceWork = make_shared<boost::asio::io_service::work>(ref(m_ioService));
+ m_impl->m_ioServiceWork.reset(new boost::asio::io_service::work(m_ioService));
}
m_ioService.run();
}
- catch (Face::ProcessEventsTimeout&) {
- // break
- m_impl->m_ioServiceWork.reset();
- }
catch (...) {
m_impl->m_ioServiceWork.reset();
m_impl->m_pendingInterestTable.clear();
@@ -405,22 +392,10 @@
if (m_transport->isConnected())
m_transport->close();
- m_impl->m_pitTimeoutCheckTimer->cancel();
- m_impl->m_processEventsTimeoutTimer->cancel();
- m_impl->m_pitTimeoutCheckTimerActive = false;
-
m_impl->m_ioServiceWork.reset();
}
void
-Face::fireProcessEventsTimeout(const boost::system::error_code& error)
-{
- if (!error) // can fire for some other reason, e.g., cancelled
- throw Face::ProcessEventsTimeout();
-}
-
-
-void
Face::onReceiveElement(const Block& blockFromDaemon)
{
const Block& block = nfd::LocalControlHeader::getPayload(blockFromDaemon);
@@ -440,14 +415,8 @@
data->getLocalControlHeader().wireDecode(blockFromDaemon);
m_impl->satisfyPendingInterests(*data);
-
- if (m_impl->m_pendingInterestTable.empty()) {
- m_impl->m_pitTimeoutCheckTimer->cancel(); // this will cause checkPitExpire invocation
- }
}
// ignore any other type
}
-
-
} // namespace ndn
diff --git a/src/face.hpp b/src/face.hpp
index a42a95f..2d76773 100644
--- a/src/face.hpp
+++ b/src/face.hpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2014 Regents of the University of California.
+ * Copyright (c) 2013-2015 Regents of the University of California.
*
* This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
*
@@ -17,8 +17,6 @@
* <http://www.gnu.org/licenses/>.
*
* See AUTHORS.md for complete list of ndn-cxx authors and contributors.
- *
- * Based on code originally written by Jeff Thompson <jefft0@remap.ucla.edu>
*/
#ifndef NDN_FACE_HPP
@@ -547,19 +545,12 @@
void
construct(shared_ptr<Transport> transport, KeyChain& keyChain);
- class ProcessEventsTimeout
- {
- };
-
void
onReceiveElement(const Block& wire);
void
asyncShutdown();
- static void
- fireProcessEventsTimeout(const boost::system::error_code& error);
-
private:
/// the IO service owned by this Face, could be null
unique_ptr<boost::asio::io_service> m_internalIoService;
diff --git a/src/transport/stream-transport.hpp b/src/transport/stream-transport.hpp
index 737842c..d7f2e49 100644
--- a/src/transport/stream-transport.hpp
+++ b/src/transport/stream-transport.hpp
@@ -127,6 +127,9 @@
}
}
+ /**
+ * @warning Must not be called directly or indirectly from within handleAsyncReceive invocation
+ */
void
resume()
{
diff --git a/src/transport/tcp-transport.cpp b/src/transport/tcp-transport.cpp
index a5b311f..c43790f 100644
--- a/src/transport/tcp-transport.cpp
+++ b/src/transport/tcp-transport.cpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2014 Regents of the University of California.
+ * Copyright (c) 2013-2015 Regents of the University of California.
*
* This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
*
@@ -41,52 +41,39 @@
TcpTransport::create(const ConfigFile& config)
{
const auto hostAndPort(getDefaultSocketHostAndPort(config));
- return make_shared<TcpTransport>(hostAndPort.first,
- hostAndPort.second);
+ return make_shared<TcpTransport>(hostAndPort.first, hostAndPort.second);
}
std::pair<std::string, std::string>
TcpTransport::getDefaultSocketHostAndPort(const ConfigFile& config)
{
const ConfigFile::Parsed& parsed = config.getParsedConfiguration();
+
std::string host = "localhost";
std::string port = "6363";
- try
- {
- const util::FaceUri uri(parsed.get<std::string>("transport"));
+ try {
+ const util::FaceUri uri(parsed.get<std::string>("transport", "tcp://" + host));
- const std::string scheme = uri.getScheme();
- if (scheme != "tcp" && scheme != "tcp4" && scheme != "tcp6")
- {
- throw Transport::Error("Cannot create TcpTransport from \"" +
- scheme + "\" URI");
- }
-
- if (!uri.getHost().empty())
- {
- host = uri.getHost();
- }
-
- if (!uri.getPort().empty())
- {
- port = uri.getPort();
- }
- }
- catch (const boost::property_tree::ptree_bad_path& error)
- {
- // no transport specified, use default host and port
- }
- catch (const boost::property_tree::ptree_bad_data& error)
- {
- throw ConfigFile::Error(error.what());
- }
- catch (const util::FaceUri::Error& error)
- {
- throw ConfigFile::Error(error.what());
+ const std::string scheme = uri.getScheme();
+ if (scheme != "tcp" && scheme != "tcp4" && scheme != "tcp6") {
+ throw Transport::Error("Cannot create TcpTransport from \"" +
+ scheme + "\" URI");
}
- return std::make_pair(host, port);
+ if (!uri.getHost().empty()) {
+ host = uri.getHost();
+ }
+
+ if (!uri.getPort().empty()) {
+ port = uri.getPort();
+ }
+ }
+ catch (const util::FaceUri::Error& error) {
+ throw ConfigFile::Error(error.what());
+ }
+
+ return {host, port};
}
void
diff --git a/tests/integrated/face.cpp b/tests/integrated/face.cpp
index 2666c48..b166edf 100644
--- a/tests/integrated/face.cpp
+++ b/tests/integrated/face.cpp
@@ -109,7 +109,7 @@
void
terminate(Face& face)
{
- face.shutdown();
+ face.getIoService().stop();
}
uint32_t nData;
@@ -191,7 +191,7 @@
Face face;
Face face2(face.getIoService());
Scheduler scheduler(face.getIoService());
- scheduler.scheduleEvent(time::milliseconds(300),
+ scheduler.scheduleEvent(time::seconds(4),
bind(&FacesFixture::terminate, this, ref(face)));
regPrefixId = face.setInterestFilter("/Hello/World",
@@ -248,7 +248,7 @@
Face face;
Face face2(face.getIoService());
Scheduler scheduler(face.getIoService());
- scheduler.scheduleEvent(time::seconds(1),
+ scheduler.scheduleEvent(time::seconds(4),
bind(&FacesFixture::terminate, this, ref(face)));
regPrefixId = face.setInterestFilter(InterestFilter("/Hello/World", "<><b><c>?"),
@@ -269,7 +269,7 @@
Face face;
Face face2(face.getIoService());
Scheduler scheduler(face.getIoService());
- scheduler.scheduleEvent(time::seconds(2),
+ scheduler.scheduleEvent(time::seconds(4),
bind(&FacesFixture::terminate, this, ref(face)));
regPrefixId = face.setInterestFilter(InterestFilter("/Hello/World", "<><b><c>?"),
@@ -308,7 +308,7 @@
void
checkPrefix(bool doesExist)
{
- int result = std::system("nfd-status | grep /Hello/World >/dev/null");
+ int result = std::system("nfd-status -r | grep /Hello/World >/dev/null");
if (doesExist) {
BOOST_CHECK_EQUAL(result, 0);
@@ -323,7 +323,7 @@
{
Face face;
Scheduler scheduler(face.getIoService());
- scheduler.scheduleEvent(time::seconds(2),
+ scheduler.scheduleEvent(time::seconds(4),
bind(&FacesFixture::terminate, this, ref(face)));
regPrefixId = face.setInterestFilter(InterestFilter("/Hello/World"),
@@ -340,7 +340,7 @@
&face,
regPrefixId)); // shouldn't match
- scheduler.scheduleEvent(time::milliseconds(1500),
+ scheduler.scheduleEvent(time::milliseconds(2000),
bind(&FacesFixture2::checkPrefix, this, false));
BOOST_REQUIRE_NO_THROW(face.processEvents());
@@ -402,7 +402,7 @@
static_cast<UnregisterPrefixSuccessCallback>(bind(&FacesFixture3::onUnregSucceeded, this)),
static_cast<UnregisterPrefixFailureCallback>(bind(&FacesFixture3::onUnregFailed, this))));
- scheduler.scheduleEvent(time::milliseconds(1500),
+ scheduler.scheduleEvent(time::milliseconds(2500),
bind(&FacesFixture2::checkPrefix, this, false));
BOOST_REQUIRE_NO_THROW(face.processEvents());
diff --git a/tests/unit-tests/face.t.cpp b/tests/unit-tests/face.t.cpp
index 3ded5ee..19e6739 100644
--- a/tests/unit-tests/face.t.cpp
+++ b/tests/unit-tests/face.t.cpp
@@ -70,16 +70,42 @@
BOOST_FAIL("Unexpected timeout");
}));
- advanceClocks(time::milliseconds(10));
+ advanceClocks(time::milliseconds(1), 40);
face->receive(*util::makeData("/Bye/World/!"));
face->receive(*util::makeData("/Hello/World/!"));
- advanceClocks(time::milliseconds(10), 100);
+ advanceClocks(time::milliseconds(1), 100);
BOOST_CHECK_EQUAL(nData, 1);
BOOST_CHECK_EQUAL(face->sentInterests.size(), 1);
BOOST_CHECK_EQUAL(face->sentDatas.size(), 0);
+
+ face->expressInterest(Interest("/Hello/World/!", time::milliseconds(50)),
+ [&] (const Interest& i, const Data& d) {
+ BOOST_CHECK(i.getName().isPrefixOf(d.getName()));
+ ++nData;
+ },
+ bind([] {
+ BOOST_FAIL("Unexpected timeout");
+ }));
+ advanceClocks(time::milliseconds(1), 40);
+ face->receive(*util::makeData("/Hello/World/!/1/xxxxx"));
+
+ advanceClocks(time::milliseconds(1), 100);
+
+ BOOST_CHECK_EQUAL(nData, 2);
+ BOOST_CHECK_EQUAL(face->sentInterests.size(), 2);
+ BOOST_CHECK_EQUAL(face->sentDatas.size(), 0);
+
+ size_t nTimeouts = 0;
+ face->expressInterest(Interest("/Hello/World/!/2", time::milliseconds(50)),
+ bind([]{}),
+ bind([&nTimeouts] {
+ ++nTimeouts;
+ }));
+ advanceClocks(time::milliseconds(10), 100);
+ BOOST_CHECK_EQUAL(nTimeouts, 1);
}
BOOST_AUTO_TEST_CASE(ExpressInterestTimeout)
@@ -367,7 +393,6 @@
Interest i("/Hello/World");
i.setNextHopFaceId(1000);
i.setIncomingFaceId(2000);
- i.getLocalControlHeader().setCachingPolicy(nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
face->expressInterest(i, bind([]{}), bind([]{}));
advanceClocks(time::milliseconds(10));
@@ -376,7 +401,6 @@
// only NextHopFaceId is allowed to go out
BOOST_CHECK(face->sentInterests[0].getLocalControlHeader().hasNextHopFaceId());
BOOST_CHECK(!face->sentInterests[0].getLocalControlHeader().hasIncomingFaceId());
- BOOST_CHECK(!face->sentInterests[0].getLocalControlHeader().hasCachingPolicy());
BOOST_CHECK_EQUAL(face->sentInterests[0].getNextHopFaceId(), 1000);
}
@@ -386,11 +410,8 @@
[] (const InterestFilter&, const Interest& i) {
BOOST_CHECK(i.getLocalControlHeader().hasNextHopFaceId());
BOOST_CHECK(i.getLocalControlHeader().hasIncomingFaceId());
- BOOST_CHECK(i.getLocalControlHeader().hasCachingPolicy());
BOOST_CHECK_EQUAL(i.getNextHopFaceId(), 1000);
BOOST_CHECK_EQUAL(i.getIncomingFaceId(), 2000);
- BOOST_CHECK_EQUAL(i.getLocalControlHeader().getCachingPolicy(),
- nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
},
bind([]{}),
bind([] {
@@ -401,7 +422,6 @@
Interest i("/Hello/World/!");
i.setNextHopFaceId(1000);
i.setIncomingFaceId(2000);
- i.getLocalControlHeader().setCachingPolicy(nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
face->receive(i);
advanceClocks(time::milliseconds(10));
@@ -411,8 +431,8 @@
{
shared_ptr<Data> d = util::makeData("/Bye/World/!");
d->setIncomingFaceId(2000);
- d->getLocalControlHeader().setNextHopFaceId(1000);
- d->setCachingPolicy(nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
+ d->getLocalControlHeader().setNextHopFaceId(1000); // setNextHopFaceId is intentionally
+ // not exposed directly
face->put(*d);
advanceClocks(time::milliseconds(10));
@@ -420,9 +440,6 @@
BOOST_REQUIRE_EQUAL(face->sentDatas.size(), 1);
BOOST_CHECK(!face->sentDatas[0].getLocalControlHeader().hasNextHopFaceId());
BOOST_CHECK(!face->sentDatas[0].getLocalControlHeader().hasIncomingFaceId());
- BOOST_CHECK(face->sentDatas[0].getLocalControlHeader().hasCachingPolicy());
- BOOST_CHECK_EQUAL(face->sentDatas[0].getCachingPolicy(),
- nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
}
BOOST_AUTO_TEST_CASE(ReceiveDataWithLocalControlHeader)
@@ -431,9 +448,6 @@
[&] (const Interest& i, const Data& d) {
BOOST_CHECK(d.getLocalControlHeader().hasNextHopFaceId());
BOOST_CHECK(d.getLocalControlHeader().hasIncomingFaceId());
- BOOST_CHECK(d.getLocalControlHeader().hasCachingPolicy());
- BOOST_CHECK_EQUAL(d.getCachingPolicy(),
- nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
BOOST_CHECK_EQUAL(d.getIncomingFaceId(), 2000);
BOOST_CHECK_EQUAL(d.getLocalControlHeader().getNextHopFaceId(), 1000);
},
@@ -445,13 +459,25 @@
shared_ptr<Data> d = util::makeData("/Hello/World/!");
d->setIncomingFaceId(2000);
- d->getLocalControlHeader().setNextHopFaceId(1000);
- d->setCachingPolicy(nfd::LocalControlHeader::CachingPolicy::NO_CACHE);
+ d->getLocalControlHeader().setNextHopFaceId(1000); // setNextHopFaceId is intentionally
+ // not exposed directly
face->receive(*d);
advanceClocks(time::milliseconds(10), 100);
}
+BOOST_AUTO_TEST_CASE(DestructionWithoutCancellingPendingInterests) // Bug #2518
+{
+ face->expressInterest(Interest("/Hello/World", time::milliseconds(50)),
+ bind([]{}), bind([]{}));
+ advanceClocks(time::milliseconds(10), 10);
+
+ face.reset();
+
+ advanceClocks(time::milliseconds(10), 10);
+ // should not segfault
+}
+
BOOST_AUTO_TEST_SUITE_END()
} // tests