From 15fa994bf663a1a4ff8b292bfb64fb07114b5a1f Mon Sep 17 00:00:00 2001 From: Davide Pesavento Date: Thu, 15 Dec 2022 04:00:22 -0500 Subject: [PATCH] transport: replace a bunch of bools with a proper state machine Change-Id: I42f74aebfd714cd8b3be7fcb3068228935482366 --- ndn-cxx/face.cpp | 2 +- ndn-cxx/impl/face-impl.hpp | 4 +- .../detail/stream-transport-impl.hpp | 35 +++++++---------- .../stream-transport-with-resolver-impl.hpp | 4 +- ndn-cxx/transport/transport.hpp | 38 +++++++++++-------- 5 files changed, 40 insertions(+), 43 deletions(-) diff --git a/ndn-cxx/face.cpp b/ndn-cxx/face.cpp index ed340e57..4a197011 100644 --- a/ndn-cxx/face.cpp +++ b/ndn-cxx/face.cpp @@ -292,7 +292,7 @@ Face::shutdown() { IO_CAPTURE_WEAK_IMPL(post) { impl->shutdown(); - if (m_transport->isConnected()) + if (m_transport->getState() != Transport::State::CLOSED) m_transport->close(); } IO_CAPTURE_WEAK_IMPL_END } diff --git a/ndn-cxx/impl/face-impl.hpp b/ndn-cxx/impl/face-impl.hpp index 1f28ca75..239885da 100644 --- a/ndn-cxx/impl/face-impl.hpp +++ b/ndn-cxx/impl/face-impl.hpp @@ -304,12 +304,12 @@ public: // IO routine void ensureConnected(bool wantResume) { - if (!m_face.m_transport->isConnected()) { + if (m_face.m_transport->getState() == Transport::State::CLOSED) { m_face.m_transport->connect(m_face.getIoService(), [this] (const Block& wire) { m_face.onReceiveElement(wire); }); } - if (wantResume && !m_face.m_transport->isReceiving()) { + if (wantResume) { m_face.m_transport->resume(); } } diff --git a/ndn-cxx/transport/detail/stream-transport-impl.hpp b/ndn-cxx/transport/detail/stream-transport-impl.hpp index f11a6229..4c52005d 100644 --- a/ndn-cxx/transport/detail/stream-transport-impl.hpp +++ b/ndn-cxx/transport/detail/stream-transport-impl.hpp @@ -55,10 +55,10 @@ public: void connect(const typename Protocol::endpoint& endpoint) { - if (m_isConnecting) { + if (m_transport.getState() == Transport::State::CONNECTING) { return; } - m_isConnecting = true; + m_transport.setState(Transport::State::CONNECTING); // Wait at most 4 seconds to connect /// @todo Decide whether this number should be configurable @@ -76,38 +76,30 @@ public: void close() { - m_isConnecting = false; + m_transport.setState(Transport::State::CLOSED); boost::system::error_code error; // to silently ignore all errors m_connectTimer.cancel(error); m_socket.cancel(error); m_socket.close(error); - m_transport.m_isConnected = false; - m_transport.m_isReceiving = false; TransmissionQueue{}.swap(m_transmissionQueue); // clear the queue } void pause() { - if (m_isConnecting) - return; - - if (m_transport.m_isReceiving) { - m_transport.m_isReceiving = false; + if (m_transport.getState() == Transport::State::RUNNING) { m_socket.cancel(); + m_transport.setState(Transport::State::PAUSED); } } void resume() { - if (m_isConnecting) - return; - - if (!m_transport.m_isReceiving) { - m_transport.m_isReceiving = true; + if (m_transport.getState() == Transport::State::PAUSED) { + m_transport.setState(Transport::State::RUNNING); m_inputBufferSize = 0; asyncReceive(); } @@ -118,7 +110,9 @@ public: { m_transmissionQueue.push(block); - if (m_transport.m_isConnected && m_transmissionQueue.size() == 1) { + if (m_transport.getState() != Transport::State::CLOSED && + m_transport.getState() != Transport::State::CONNECTING && + m_transmissionQueue.size() == 1) { asyncWrite(); } // if not connected or there's another transmission in progress (m_transmissionQueue.size() > 1), @@ -129,16 +123,14 @@ protected: void connectHandler(const boost::system::error_code& error) { - m_isConnecting = false; m_connectTimer.cancel(); if (error) { - m_transport.m_isConnected = false; m_transport.close(); NDN_THROW(Transport::Error(error, "error while connecting to the forwarder")); } - m_transport.m_isConnected = true; + m_transport.setState(Transport::State::PAUSED); if (!m_transmissionQueue.empty()) { resume(); @@ -172,8 +164,8 @@ protected: NDN_THROW(Transport::Error(error, "error while writing data to socket")); } - if (!m_transport.m_isConnected) { - return; // queue has been already cleared + if (m_transport.getState() == Transport::State::CLOSED) { + return; // queue has already been cleared } BOOST_ASSERT(!m_transmissionQueue.empty()); @@ -249,7 +241,6 @@ protected: size_t m_inputBufferSize = 0; TransmissionQueue m_transmissionQueue; boost::asio::steady_timer m_connectTimer; - bool m_isConnecting = false; }; } // namespace detail diff --git a/ndn-cxx/transport/detail/stream-transport-with-resolver-impl.hpp b/ndn-cxx/transport/detail/stream-transport-with-resolver-impl.hpp index ded87f6c..61c7d279 100644 --- a/ndn-cxx/transport/detail/stream-transport-with-resolver-impl.hpp +++ b/ndn-cxx/transport/detail/stream-transport-with-resolver-impl.hpp @@ -42,10 +42,10 @@ public: void connect(const typename Protocol::resolver::query& query) { - if (this->m_isConnecting) { + if (this->m_transport.getState() == Transport::State::CONNECTING) { return; } - this->m_isConnecting = true; + this->m_transport.setState(Transport::State::CONNECTING); // Wait at most 4 seconds to connect /// @todo Decide whether this number should be configurable diff --git a/ndn-cxx/transport/transport.hpp b/ndn-cxx/transport/transport.hpp index 6fd43748..2ef4b3d3 100644 --- a/ndn-cxx/transport/transport.hpp +++ b/ndn-cxx/transport/transport.hpp @@ -30,7 +30,8 @@ namespace ndn { -/** \brief Provides TLV-block delivery service. +/** + * \brief Provides a "TLV-oriented" delivery service. */ class Transport : noncopyable { @@ -43,9 +44,16 @@ public: Error(const boost::system::error_code& code, const std::string& msg); }; - using ReceiveCallback = std::function; - using ErrorCallback = std::function; + enum class State { + CLOSED, + CONNECTING, + RUNNING, + PAUSED, + }; + using ReceiveCallback = std::function; + +public: virtual ~Transport() = default; @@ -89,29 +97,27 @@ public: resume() = 0; /** - * \brief Return whether the transport is connected. + * \brief Return the current state of the transport. */ - bool - isConnected() const noexcept + State + getState() const noexcept { - return m_isConnected; + return m_state; } - /** - * \retval true incoming packets are expected, the receive callback will be invoked - * \retval false incoming packets are not expected, the receive callback will not be invoked - */ - bool - isReceiving() const noexcept +protected: + void + setState(State state) noexcept { - return m_isReceiving; + m_state = state; } protected: boost::asio::io_service* m_ioService = nullptr; ReceiveCallback m_receiveCallback; - bool m_isConnected = false; - bool m_isReceiving = false; + +private: + State m_state = State::CLOSED; }; } // namespace ndn