face: Implementing close operation and many related fixes in TcpFace and TcpChannel
Change-Id: Ib6b751e80454e149bf94f3867663d5e705cbf4a0
refs: #1250, #1248
diff --git a/daemon/face/face.hpp b/daemon/face/face.hpp
index ed78504..e2aa62d 100644
--- a/daemon/face/face.hpp
+++ b/daemon/face/face.hpp
@@ -27,6 +27,14 @@
class Face : noncopyable, public enable_shared_from_this<Face>
{
public:
+ /**
+ * \brief Face-related error
+ */
+ struct Error : public std::runtime_error
+ {
+ Error(const std::string& what) : std::runtime_error(what) {}
+ };
+
Face();
virtual
@@ -51,6 +59,15 @@
/// send a Data
virtual void
sendData(const Data& data) = 0;
+
+ /**
+ * \brief Close the face
+ *
+ * This terminates all communication on the face and cause
+ * onFail() method event to be invoked
+ */
+ virtual void
+ close() = 0;
/** \brief Get whether underlying communicate is up
* In this base class this property is always true.
diff --git a/daemon/face/stream-face.hpp b/daemon/face/stream-face.hpp
index 529e3ef..65ca82e 100644
--- a/daemon/face/stream-face.hpp
+++ b/daemon/face/stream-face.hpp
@@ -17,8 +17,24 @@
public:
typedef T protocol;
+ /**
+ * \brief Create instance of StreamFace
+ */
StreamFace(const shared_ptr<typename protocol::socket>& socket);
+ virtual
+ ~StreamFace();
+
+ // from Face
+ virtual void
+ sendInterest(const Interest& interest);
+
+ virtual void
+ sendData(const Data& data);
+
+ virtual void
+ close();
+
protected:
void
handleSend(const boost::system::error_code& error,
@@ -28,17 +44,25 @@
handleReceive(const boost::system::error_code& error,
std::size_t bytes_recvd);
+ void
+ keepFaceAliveUntilAllHandlersExecuted(const shared_ptr<Face>& face);
+
protected:
shared_ptr<typename protocol::socket> m_socket;
-
+
private:
uint8_t m_inputBuffer[MAX_NDN_PACKET_SIZE];
std::size_t m_inputBufferSize;
+#ifdef _DEBUG
+ typename protocol::endpoint m_localEndpoint;
+#endif
+
NFD_LOG_INCLASS_DECLARE();
};
-NFD_LOG_INCLASS_TEMPLATE_DEFINE(StreamFace, "StreamFace");
+// All inherited classes must use
+// NFD_LOG_INCLASS_TEMPLATE_SPECIALIZATION_DEFINE(StreamFace, <specialization-parameter>, "Name");
template <class T>
inline
@@ -49,6 +73,58 @@
bind(&StreamFace<T>::handleReceive, this, _1, _2));
}
+template <class T>
+inline
+StreamFace<T>::~StreamFace()
+{
+}
+
+
+template <class T>
+inline void
+StreamFace<T>::sendInterest(const Interest& interest)
+{
+ m_socket->async_send(boost::asio::buffer(interest.wireEncode().wire(),
+ interest.wireEncode().size()),
+ bind(&StreamFace<T>::handleSend, this, _1, interest.wireEncode()));
+
+ // anything else should be done here?
+}
+
+template <class T>
+inline void
+StreamFace<T>::sendData(const Data& data)
+{
+ m_socket->async_send(boost::asio::buffer(data.wireEncode().wire(),
+ data.wireEncode().size()),
+ bind(&StreamFace<T>::handleSend, this, _1, data.wireEncode()));
+
+ // anything else should be done here?
+}
+
+template <class T>
+inline void
+StreamFace<T>::close()
+{
+ if (!m_socket->is_open())
+ return;
+
+ NFD_LOG_INFO("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Close connection");
+
+
+ boost::asio::io_service& io = m_socket->get_io_service();
+ m_socket->close();
+ // after this, handleSend/handleReceive will be called with set error code
+
+ // ensure that if that Face object is alive at least until all pending
+ // methods are dispatched
+ io.post(bind(&StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted,
+ this, this->shared_from_this()));
+
+ onFail("Close connection");
+}
template <class T>
inline void
@@ -59,14 +135,43 @@
if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
return;
- NFD_LOG_WARN("[id:" << this->getId()
- << ",endpoint:" << m_socket->local_endpoint()
- << "] Send operation failed, closing socket: "
- << error.category().message(error.value()));
+ if (!m_socket->is_open())
+ {
+ onFail("Connection closed");
+ return;
+ }
- onFail("Send operation failed, closing socket: " +
- error.category().message(error.value()));
+ if (error == boost::asio::error::eof)
+ {
+ NFD_LOG_INFO("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Connection closed");
+ }
+ else
+ {
+ NFD_LOG_WARN("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Send operation failed, closing socket: "
+ << error.category().message(error.value()));
+ }
+
+ boost::asio::io_service& io = m_socket->get_io_service();
m_socket->close();
+
+ // ensure that if that Face object is alive at least until all pending
+ // methods are dispatched
+ io.post(bind(&StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted,
+ this, this->shared_from_this()));
+
+ if (error == boost::asio::error::eof)
+ {
+ onFail("Connection closed");
+ }
+ else
+ {
+ onFail("Send operation failed, closing socket: " +
+ error.category().message(error.value()));
+ }
return;
}
@@ -81,25 +186,55 @@
StreamFace<T>::handleReceive(const boost::system::error_code& error,
std::size_t bytes_recvd)
{
- NFD_LOG_TRACE("[id:" << this->getId()
- << ",endpoint:" << m_socket->local_endpoint()
- << "] Received: " << bytes_recvd << " bytes");
-
if (error || bytes_recvd == 0) {
if (error == boost::system::errc::operation_canceled) // when socket is closed by someone
return;
- NFD_LOG_WARN("[id:" << this->getId()
- << ",endpoint:" << m_socket->local_endpoint()
- << "] Receive operation failed: "
- << error.category().message(error.value()));
+ // this should be unnecessary, but just in case
+ if (!m_socket->is_open())
+ {
+ onFail("Connection closed");
+ return;
+ }
+
+ if (error == boost::asio::error::eof)
+ {
+ NFD_LOG_INFO("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Connection closed");
+ }
+ else
+ {
+ NFD_LOG_WARN("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Receive operation failed: "
+ << error.category().message(error.value()));
+ }
- onFail("Receive operation failed, closing socket: " +
- error.category().message(error.value()));
+ boost::asio::io_service& io = m_socket->get_io_service();
m_socket->close();
+
+ // ensure that if that Face object is alive at least until all pending
+ // methods are dispatched
+ io.post(bind(&StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted,
+ this, this->shared_from_this()));
+
+ if (error == boost::asio::error::eof)
+ {
+ onFail("Connection closed");
+ }
+ else
+ {
+ onFail("Receive operation failed, closing socket: " +
+ error.category().message(error.value()));
+ }
return;
}
+ NFD_LOG_TRACE("[id:" << this->getId()
+ << ",endpoint:" << m_socket->local_endpoint()
+ << "] Received: " << bytes_recvd << " bytes");
+
m_inputBufferSize += bytes_recvd;
// do magic
@@ -150,8 +285,15 @@
<< "] Received input is invalid or too large to process, "
<< "closing down the face");
- onFail("Received input is invalid or too large to process, closing down the face");
+ boost::asio::io_service& io = m_socket->get_io_service();
m_socket->close();
+
+ // ensure that if that Face object is alive at least until all pending
+ // methods are dispatched
+ io.post(bind(&StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted,
+ this, this->shared_from_this()));
+
+ onFail("Received input is invalid or too large to process, closing down the face");
return;
}
}
@@ -175,6 +317,12 @@
bind(&StreamFace<T>::handleReceive, this, _1, _2));
}
+template <class T>
+inline void
+StreamFace<T>::keepFaceAliveUntilAllHandlersExecuted(const shared_ptr<Face>& face)
+{
+}
+
} // namespace nfd
diff --git a/daemon/face/tcp-channel.cpp b/daemon/face/tcp-channel.cpp
index 026c338..087a24c 100644
--- a/daemon/face/tcp-channel.cpp
+++ b/daemon/face/tcp-channel.cpp
@@ -111,19 +111,33 @@
onConnectFailed));
}
+size_t
+TcpChannel::size() const
+{
+ return m_channelFaces.size();
+}
void
TcpChannel::createFace(const shared_ptr<ip::tcp::socket>& socket,
const FaceCreatedCallback& onFaceCreated)
{
- shared_ptr<TcpFace> face = make_shared<TcpFace>(boost::cref(socket));
- onFaceCreated(face);
-
tcp::Endpoint remoteEndpoint = socket->remote_endpoint();
+
+ shared_ptr<TcpFace> face = make_shared<TcpFace>(boost::cref(socket));
+ face->onFail += bind(&TcpChannel::afterFaceFailed, this, remoteEndpoint);
+
+ onFaceCreated(face);
m_channelFaces[remoteEndpoint] = face;
}
void
+TcpChannel::afterFaceFailed(tcp::Endpoint &remoteEndpoint)
+{
+ NFD_LOG_DEBUG("afterFaceFailed: " << remoteEndpoint);
+ m_channelFaces.erase(remoteEndpoint);
+}
+
+void
TcpChannel::handleSuccessfulAccept(const boost::system::error_code& error,
const shared_ptr<boost::asio::ip::tcp::socket>& socket,
const FaceCreatedCallback& onFaceCreated,
diff --git a/daemon/face/tcp-channel.hpp b/daemon/face/tcp-channel.hpp
index 999f9c9..e94747f 100644
--- a/daemon/face/tcp-channel.hpp
+++ b/daemon/face/tcp-channel.hpp
@@ -52,8 +52,11 @@
/**
* \brief Enable listening on the local endpoint, accept connections,
* and create faces when remote host makes a connection
- * \param backlog The maximum length of the queue of pending incoming
- * connections
+ * \param onFaceCreated Callback to notify successful creation of the face
+ * \param onAcceptFailed Callback to notify when channel fails (accept call
+ * returns an error)
+ * \param backlog The maximum length of the queue of pending incoming
+ * connections
*/
void
listen(const FaceCreatedCallback& onFaceCreated,
@@ -85,6 +88,12 @@
const FaceCreatedCallback& onFaceCreated,
const ConnectFailedCallback& onConnectFailed,
const time::Duration& timeout = time::seconds(4));
+
+ /**
+ * \brief Get number of faces in the channel
+ */
+ size_t
+ size() const;
private:
void
@@ -92,6 +101,9 @@
const FaceCreatedCallback& onFaceCreated);
void
+ afterFaceFailed(tcp::Endpoint &endpoint);
+
+ void
handleSuccessfulAccept(const boost::system::error_code& error,
const shared_ptr<boost::asio::ip::tcp::socket>& socket,
const FaceCreatedCallback& onFaceCreated,
@@ -118,7 +130,7 @@
const FaceCreatedCallback& onFaceCreated,
const ConnectFailedCallback& onConnectFailed,
const shared_ptr<boost::asio::ip::tcp::resolver>& resolver);
-
+
private:
boost::asio::io_service& m_ioService;
tcp::Endpoint m_localEndpoint;
diff --git a/daemon/face/tcp-face.cpp b/daemon/face/tcp-face.cpp
index a609725..e867483 100644
--- a/daemon/face/tcp-face.cpp
+++ b/daemon/face/tcp-face.cpp
@@ -8,6 +8,9 @@
namespace nfd {
+// The whole purpose of this file is to specialize the logger,
+// otherwise, everything could be put into the header file.
+
NFD_LOG_INCLASS_TEMPLATE_SPECIALIZATION_DEFINE(StreamFace, TcpFace::protocol, "TcpFace");
TcpFace::TcpFace(const shared_ptr<TcpFace::protocol::socket>& socket)
@@ -15,24 +18,5 @@
{
}
-void
-TcpFace::sendInterest(const Interest& interest)
-{
- m_socket->async_send(boost::asio::buffer(interest.wireEncode().wire(),
- interest.wireEncode().size()),
- bind(&TcpFace::handleSend, this, _1, interest.wireEncode()));
-
- // anything else should be done here?
-}
-
-void
-TcpFace::sendData(const Data& data)
-{
- m_socket->async_send(boost::asio::buffer(data.wireEncode().wire(),
- data.wireEncode().size()),
- bind(&TcpFace::handleSend, this, _1, data.wireEncode()));
-
- // anything else should be done here?
-}
} // namespace nfd
diff --git a/daemon/face/tcp-face.hpp b/daemon/face/tcp-face.hpp
index cf3188c..af45c6c 100644
--- a/daemon/face/tcp-face.hpp
+++ b/daemon/face/tcp-face.hpp
@@ -22,13 +22,6 @@
typedef boost::asio::ip::tcp protocol;
TcpFace(const shared_ptr<protocol::socket>& socket);
-
- // from Face
- virtual void
- sendInterest(const Interest& interest);
-
- virtual void
- sendData(const Data& data);
};
} // namespace nfd
diff --git a/daemon/mgmt/internal-face.cpp b/daemon/mgmt/internal-face.cpp
index 7af328f..32c03a1 100644
--- a/daemon/mgmt/internal-face.cpp
+++ b/daemon/mgmt/internal-face.cpp
@@ -90,10 +90,16 @@
void
InternalFace::sendData(const Data& data)
{
-
}
void
+InternalFace::close()
+{
+ throw Error("Internal face cannot be closed");
+}
+
+
+void
InternalFace::setInterestFilter(const Name& filter,
OnInterest onInterest)
{
diff --git a/daemon/mgmt/internal-face.hpp b/daemon/mgmt/internal-face.hpp
index 9539a9e..8d5805c 100644
--- a/daemon/mgmt/internal-face.hpp
+++ b/daemon/mgmt/internal-face.hpp
@@ -15,6 +15,13 @@
class InternalFace : public Face, public AppFace
{
public:
+ /**
+ * \brief InternalFace-related error
+ */
+ struct Error : public Face::Error
+ {
+ Error(const std::string& what) : Face::Error(what) {}
+ };
InternalFace();
@@ -26,6 +33,9 @@
virtual void
sendData(const Data& data);
+ virtual void
+ close();
+
// Methods implementing AppFace interface. Do not invoke from forwarder.
virtual void