diff options
26 files changed, 330 insertions, 175 deletions
diff --git a/libtransport/src/hicn/transport/core/connector.h b/libtransport/src/hicn/transport/core/connector.h index c790f2bfb..f2bbe5dcd 100644 --- a/libtransport/src/hicn/transport/core/connector.h +++ b/libtransport/src/hicn/transport/core/connector.h @@ -64,10 +64,10 @@ class Connector { virtual void close() = 0; - virtual void enableBurst() = 0; - virtual ConnectorState state() { return state_; }; + virtual bool isConnected() { return state_ == ConnectorState::CONNECTED; } + protected: void increasePoolSize(std::size_t size = packet_pool_size); diff --git a/libtransport/src/hicn/transport/core/forwarder_interface.h b/libtransport/src/hicn/transport/core/forwarder_interface.h index b4bc26da7..8fefba8ad 100644 --- a/libtransport/src/hicn/transport/core/forwarder_interface.h +++ b/libtransport/src/hicn/transport/core/forwarder_interface.h @@ -101,15 +101,13 @@ class ForwarderInterface { connector_.send(packet.acquireMemBufReference()); } - template <typename Handler> - TRANSPORT_ALWAYS_INLINE void send(const uint8_t *packet, std::size_t len, - Handler &&packet_sent) { + TRANSPORT_ALWAYS_INLINE void send(const uint8_t *packet, std::size_t len) { // ASIO_COMPLETION_HANDLER_CHECK(Handler, packet_sent) type_check; counters_.tx_packets++; counters_.tx_bytes += len; // Perfect forwarding - connector_.send(packet, len, std::forward<Handler &&>(packet_sent)); + connector_.send(packet, len); } TRANSPORT_ALWAYS_INLINE void shutdown() { connector_.close(); } diff --git a/libtransport/src/hicn/transport/core/hicn_forwarder_interface.cc b/libtransport/src/hicn/transport/core/hicn_forwarder_interface.cc index 585d80f8c..33a37f540 100644 --- a/libtransport/src/hicn/transport/core/hicn_forwarder_interface.cc +++ b/libtransport/src/hicn/transport/core/hicn_forwarder_interface.cc @@ -61,44 +61,44 @@ void fillCommandHeader(CommandHeader *header) { header->length = 1; } -std::unique_ptr<RouteToSelfCommand> createCommandRoute( - std::unique_ptr<sockaddr> &&addr, uint8_t prefix_length) { - auto command = std::make_unique<RouteToSelfCommand>(); +RouteToSelfCommand createCommandRoute(std::unique_ptr<sockaddr> &&addr, + uint8_t prefix_length) { + RouteToSelfCommand command; // check and set IP address if (addr->sa_family == AF_INET) { - command->address_type = addr_inet; - command->address.ipv4 = ((sockaddr_in *)addr.get())->sin_addr.s_addr; + command.address_type = addr_inet; + command.address.ipv4 = ((sockaddr_in *)addr.get())->sin_addr.s_addr; } else if (addr->sa_family == AF_INET6) { - command->address_type = addr_inet6; - command->address.ipv6 = ((sockaddr_in6 *)addr.get())->sin6_addr; + command.address_type = addr_inet6; + command.address.ipv6 = ((sockaddr_in6 *)addr.get())->sin6_addr; } // Fill remaining payload fields #ifndef _WIN32 - strcpy(command->symbolic_or_connid, identifier); + strcpy(command.symbolic_or_connid, identifier); #else - strcpy_s(command->symbolic_or_connid, 16, identifier); + strcpy_s(command.symbolic_or_connid, 16, identifier); #endif - command->cost = 1; - command->len = (uint8_t)prefix_length; + command.cost = 1; + command.len = (uint8_t)prefix_length; // Allocate and fill the header - command->command_id = add_route_command; - fillCommandHeader((CommandHeader *)command.get()); + command.command_id = add_route_command; + fillCommandHeader((CommandHeader *)&command); return command; } -std::unique_ptr<DeleteSelfConnectionCommand> createCommandDeleteConnection() { - auto command = std::make_unique<DeleteSelfConnectionCommand>(); - fillCommandHeader((CommandHeader *)command.get()); - command->command_id = delete_connection_command; +DeleteSelfConnectionCommand createCommandDeleteConnection() { + DeleteSelfConnectionCommand command; + fillCommandHeader((CommandHeader *)&command); + command.command_id = delete_connection_command; #ifndef _WIN32 - strcpy(command->symbolic_or_connid, identifier); + strcpy(command.symbolic_or_connid, identifier); #else - strcpy_s(command->symbolic_or_connid, 16, identifier); + strcpy_s(command.symbolic_or_connid, 16, identifier); #endif return command; @@ -119,20 +119,15 @@ HicnForwarderInterface::~HicnForwarderInterface() {} void HicnForwarderInterface::connect(bool is_consumer) { connector_.connect(); } void HicnForwarderInterface::registerRoute(Prefix &prefix) { - auto command = - createCommandRoute(prefix.toSockaddr(), (uint8_t)prefix.getPrefixLength()) - .release(); - send((uint8_t *)command, sizeof(RouteToSelfCommand), - [command]() { delete command; }); + auto command = createCommandRoute(prefix.toSockaddr(), + (uint8_t)prefix.getPrefixLength()); + send((uint8_t *)&command, sizeof(RouteToSelfCommand)); } void HicnForwarderInterface::closeConnection() { - auto command = createCommandDeleteConnection().release(); - send((uint8_t *)command, sizeof(DeleteSelfConnectionCommand), - [this, command]() { - delete command; - connector_.close(); - }); + auto command = createCommandDeleteConnection(); + send((uint8_t *)&command, sizeof(DeleteSelfConnectionCommand)); + connector_.close(); } } // namespace core diff --git a/libtransport/src/hicn/transport/core/memif_connector.cc b/libtransport/src/hicn/transport/core/memif_connector.cc index f9695800b..a77f14839 100644 --- a/libtransport/src/hicn/transport/core/memif_connector.cc +++ b/libtransport/src/hicn/transport/core/memif_connector.cc @@ -14,6 +14,7 @@ */ #include <hicn/transport/core/memif_connector.h> +#include <hicn/transport/errors/not_implemented_exception.h> #ifdef __vpp__ @@ -69,7 +70,6 @@ MemifConnector::MemifConnector(PacketReceivedCallback &&receive_callback, tx_buf_counter_(0), is_reconnection_(false), data_available_(false), - enable_burst_(false), app_name_(app_name), socket_filename_("") { std::call_once(MemifConnector::flag_, &MemifConnector::init, this); @@ -429,8 +429,6 @@ void MemifConnector::close() { } } -void MemifConnector::enableBurst() { enable_burst_ = true; } - void MemifConnector::send(const Packet::MemBufPtr &packet) { { utils::SpinLock::Acquire locked(write_msgs_lock_); @@ -496,7 +494,9 @@ int MemifConnector::doSend() { } void MemifConnector::send(const uint8_t *packet, std::size_t len, - const PacketSentCallback &packet_sent) {} + const PacketSentCallback &packet_sent) { + throw errors::NotImplementedException(); +} } // end namespace core diff --git a/libtransport/src/hicn/transport/core/memif_connector.h b/libtransport/src/hicn/transport/core/memif_connector.h index 057df37e4..ef100e3fb 100644 --- a/libtransport/src/hicn/transport/core/memif_connector.h +++ b/libtransport/src/hicn/transport/core/memif_connector.h @@ -65,10 +65,6 @@ class MemifConnector : public Connector { void connect(uint32_t memif_id, long memif_mode); - // void runEventsLoop(); - - void enableBurst() override; - TRANSPORT_ALWAYS_INLINE uint32_t getMemifId() { return memif_id_; }; private: @@ -120,7 +116,6 @@ class MemifConnector : public Connector { PacketRing input_buffer_; bool is_reconnection_; bool data_available_; - bool enable_burst_; uint32_t memif_id_; uint8_t memif_mode_; std::string app_name_; diff --git a/libtransport/src/hicn/transport/core/portal.h b/libtransport/src/hicn/transport/core/portal.h index 07f84075e..3ea37c938 100644 --- a/libtransport/src/hicn/transport/core/portal.h +++ b/libtransport/src/hicn/transport/core/portal.h @@ -260,6 +260,26 @@ class BasicBindConfig { using BindConfig = BasicBindConfig<Prefix>; +/** + * Portal is a opaque class which is used for sending/receiving interest/data + * packets over multiple kind of connector. The connector itself is defined by + * the template ForwarderInt, which is resolved at compile time. It is then not + * possible to decide at runtime what the connector will be. + * + * The tasks performed by portal are the following: + * - Sending/Receiving Interest packets + * - Sending/Receiving Data packets + * - Set timers (one per interest), in order to trigger events if an interest is + * not satisfied + * - Register a producer prefix to the local forwarder + * + * The way of working of portal is event-based, which means that data and + * interests are sent/received in a asynchronous manner, and the notifications + * are performed through callbacks. + * + * The portal class is not thread safe, appropriate locking is required by the + * users of this class. + */ template <typename ForwarderInt> class Portal { static_assert( @@ -269,12 +289,20 @@ class Portal { "ForwarderInt must inherit from ForwarderInterface!"); public: + /** + * Consumer callback is an abstract class containing two methods to be + * implemented by a consumer application. + */ class ConsumerCallback { public: virtual void onContentObject(Interest::Ptr &&i, ContentObject::Ptr &&c) = 0; virtual void onTimeout(Interest::Ptr &&i) = 0; }; + /** + * Producer callback is an abstract class containing two methods to be + * implemented by a producer application. + */ class ProducerCallback { public: virtual void onInterest(Interest::Ptr &&i) = 0; @@ -287,33 +315,65 @@ class Portal { app_name_("libtransport_application"), consumer_callback_(nullptr), producer_callback_(nullptr), + packet_pool_(io_service), connector_(std::bind(&Portal::processIncomingMessages, this, std::placeholders::_1), std::bind(&Portal::setLocalRoutes, this), io_service_, app_name_), - forwarder_interface_(connector_), - packet_pool_(io_service) {} + forwarder_interface_(connector_) {} + /** + * Set the consumer callback. + * + * @param consumer_callback - The pointer to the ConsumerCallback object. + */ void setConsumerCallback(ConsumerCallback *consumer_callback) { consumer_callback_ = consumer_callback; } + /** + * Set the producer callback. + * + * @param producer_callback - The pointer to the ProducerCallback object. + */ void setProducerCallback(ProducerCallback *producer_callback) { producer_callback_ = producer_callback; } + /** + * Specify the output interface to use. This method will be useful in a future + * scenario where the library will be able to forward packets without + * connecting to a local forwarder. Now it is not used. + * + * @param output_interface - The output interface to use for + * forwarding/receiving packets. + */ TRANSPORT_ALWAYS_INLINE void setOutputInterface( const std::string &output_interface) { forwarder_interface_.setOutputInterface(output_interface); } + /** + * Connect the transport to the local hicn forwarder. + * + * @param is_consumer - Boolean specifying if the application on top of portal + * is a consumer or a producer. + */ TRANSPORT_ALWAYS_INLINE void connect(bool is_consumer = true) { pending_interest_hash_table_.reserve(portal_details::pool_size); forwarder_interface_.connect(is_consumer); } + /** + * Destructor. + */ ~Portal() { killConnection(); } + /** + * Check if there is already a pending interest for a given name. + * + * @param name - The interest name. + */ TRANSPORT_ALWAYS_INLINE bool interestIsPending(const Name &name) { auto it = pending_interest_hash_table_.find(name.getHash32() + name.getSuffix()); @@ -324,6 +384,20 @@ class Portal { return false; } + /** + * Send an interest through to the local forwarder. + * + * @param interest - The pointer to the interest. The ownership of the + * interest is transferred by the caller to portal. + * + * @param on_content_object_callback - If the caller wishes to use a different + * callback to be called for this interest, it can set this parameter. + * Otherwise ConsumerCallback::onContentObject will be used. + * + * @param on_interest_timeout_callback - If the caller wishes to use a + * different callback to be called for this interest, it can set this + * parameter. Otherwise ConsumerCallback::onTimeout will be used. + */ TRANSPORT_ALWAYS_INLINE void sendInterest( Interest::Ptr &&interest, OnContentObjectCallback &&on_content_object_callback = UNSET_CALLBACK, @@ -347,6 +421,14 @@ class Portal { std::make_pair(hash, std::move(pending_interest))); } + /** + * Handler fot the timer set when the interest is sent. + * + * @param ec - Error code which says whether the timer expired or has been + * canceled upon data packet reception. + * + * @param hash - The index of the interest in the pending interest hash table. + */ TRANSPORT_ALWAYS_INLINE void timerHandler(const std::error_code &ec, uint32_t hash) { bool is_stopped = io_service_.stopped(); @@ -370,34 +452,59 @@ class Portal { } } + /** + * Register a producer name to the local forwarder and optionally set the + * content store size in a per-face manner. + * + * @param config - The configuration for the local forwarder binding. + */ TRANSPORT_ALWAYS_INLINE void bind(const BindConfig &config) { - connector_.enableBurst(); forwarder_interface_.setContentStoreSize(config.csReserved()); served_namespaces_.push_back(config.prefix()); - registerRoute(served_namespaces_.back()); + + setLocalRoutes(); } + /** + * Start the event loop. This function blocks here and calls the callback set + * by the application upon interest/data received or timeout. + */ TRANSPORT_ALWAYS_INLINE void runEventsLoop() { if (io_service_.stopped()) { io_service_.reset(); // ensure that run()/poll() will do some work } - this->io_service_.run(); + io_service_.run(); } + /** + * Run one event and return. + */ TRANSPORT_ALWAYS_INLINE void runOneEvent() { if (io_service_.stopped()) { io_service_.reset(); // ensure that run()/poll() will do some work } - this->io_service_.run_one(); + io_service_.run_one(); } + /** + * Send a data packet to the local forwarder. As opposite to sendInterest, the + * ownership of the content object is not transferred to the portal. + * + * @param content_object - The data packet. + */ TRANSPORT_ALWAYS_INLINE void sendContentObject( ContentObject &content_object) { forwarder_interface_.send(content_object); } + /** + * Stop the event loop, canceling all the pending events in the event queue. + * + * Beware that stopping the event loop DOES NOT disconnect the transport from + * the local forwarder, the connector underneath will stay connected. + */ TRANSPORT_ALWAYS_INLINE void stopEventsLoop() { if (!io_service_.stopped()) { io_service_.dispatch([this]() { @@ -407,32 +514,59 @@ class Portal { } } + /** + * Disconnect the transport from the local forwarder. + */ TRANSPORT_ALWAYS_INLINE void killConnection() { forwarder_interface_.closeConnection(); } + /** + * Clear the pending interest hash table. + */ TRANSPORT_ALWAYS_INLINE void clear() { - for (auto &pend_interest : pending_interest_hash_table_) { - pend_interest.second->cancelTimer(); + if (!io_service_.stopped()) { + io_service_.dispatch(std::bind(&Portal::doClear, this)); + } else { + doClear(); } - - pending_interest_hash_table_.clear(); } + /** + * Get a reference to the io_service object. + */ TRANSPORT_ALWAYS_INLINE asio::io_service &getIoService() { return io_service_; } - TRANSPORT_ALWAYS_INLINE std::size_t getPITSize() { - connector_.state(); - return pending_interest_hash_table_.size(); - } - + /** + * Register a route to the local forwarder. + */ TRANSPORT_ALWAYS_INLINE void registerRoute(Prefix &prefix) { - forwarder_interface_.registerRoute(prefix); + served_namespaces_.push_back(prefix); + if (connector_.isConnected()) { + forwarder_interface_.registerRoute(prefix); + } } private: + /** + * Clear the pending interest hash table. + */ + TRANSPORT_ALWAYS_INLINE void doClear() { + for (auto &pend_interest : pending_interest_hash_table_) { + pend_interest.second->cancelTimer(); + } + + pending_interest_hash_table_.clear(); + } + + /** + * Callback called by the underlying connector upon reception of a packet from + * the local forwarder. + * + * @param packet_buffer - The bytes of the packet. + */ TRANSPORT_ALWAYS_INLINE void processIncomingMessages( Packet::MemBufPtr &&packet_buffer) { bool is_stopped = io_service_.stopped(); @@ -463,9 +597,16 @@ class Portal { } } + /** + * Callback called by the transport upon connection to the local forwarder. + * It register the prefixes in the served_namespaces_ list to the local + * forwarder. + */ TRANSPORT_ALWAYS_INLINE void setLocalRoutes() { - for (auto &name : served_namespaces_) { - registerRoute(name); + for (auto &prefix : served_namespaces_) { + if (connector_.isConnected()) { + forwarder_interface_.registerRoute(prefix); + } } } @@ -476,6 +617,14 @@ class Portal { } } + /** + * Process a content object: + * - Check if the data packet was effectively requested by portal + * - Delete its timer + * - Pass packet to application + * + * @param content_object - The data packet + */ TRANSPORT_ALWAYS_INLINE void processContentObject( ContentObject::Ptr &&content_object) { uint32_t hash = content_object->getName().getHash32() + @@ -500,6 +649,11 @@ class Portal { } } + /** + * Process a control message. Control messages are different depending on the + * connector, then the forwarder_interface will do the job of understanding + * them. + */ TRANSPORT_ALWAYS_INLINE void processControlMessage( Packet::MemBufPtr &&packet_buffer) { forwarder_interface_.processControlMessageReply(std::move(packet_buffer)); @@ -512,19 +666,18 @@ class Portal { std::string app_name_; PendingInterestHashTable pending_interest_hash_table_; + std::list<Prefix> served_namespaces_; ConsumerCallback *consumer_callback_; ProducerCallback *producer_callback_; - typename ForwarderInt::ConnectorType connector_; - ForwarderInt forwarder_interface_; - - std::list<Prefix> served_namespaces_; portal_details::Pool packet_pool_; - portal_details::HandlerMemory async_callback_memory_; + + typename ForwarderInt::ConnectorType connector_; + ForwarderInt forwarder_interface_; }; -} // end namespace core +} // namespace core } // end namespace transport diff --git a/libtransport/src/hicn/transport/core/raw_socket_connector.cc b/libtransport/src/hicn/transport/core/raw_socket_connector.cc index 12cc4e0fa..0e1743671 100644 --- a/libtransport/src/hicn/transport/core/raw_socket_connector.cc +++ b/libtransport/src/hicn/transport/core/raw_socket_connector.cc @@ -115,11 +115,17 @@ void RawSocketConnector::connect(const std::string &interface_name, void RawSocketConnector::send(const uint8_t *packet, std::size_t len, const PacketSentCallback &packet_sent) { - // asio::async_write(socket_, asio::buffer(packet, len), - // [packet_sent] (std::error_code ec, - // std::size_t /*length*/) { - // packet_sent(); - // }); + if (packet_sent != 0) { + socket_.async_send( + asio::buffer(packet, len), + [packet_sent](std::error_code ec, std::size_t /*length*/) { + packet_sent(); + }); + } else { + if (state_ == ConnectorState::CONNECTED) { + socket_.send(asio::buffer(packet, len)); + } + } } void RawSocketConnector::send(const Packet::MemBufPtr &packet) { @@ -191,8 +197,6 @@ void RawSocketConnector::doConnect() { socket_.bind(raw_endpoint(&link_layer_address_, sizeof(link_layer_address_))); } -void RawSocketConnector::enableBurst() { return; } - } // end namespace core } // end namespace transport diff --git a/libtransport/src/hicn/transport/core/raw_socket_connector.h b/libtransport/src/hicn/transport/core/raw_socket_connector.h index a5474f7f8..fe9ceb227 100644 --- a/libtransport/src/hicn/transport/core/raw_socket_connector.h +++ b/libtransport/src/hicn/transport/core/raw_socket_connector.h @@ -49,8 +49,6 @@ class RawSocketConnector : public Connector { void close() override; - void enableBurst() override; - void connect(const std::string &interface_name, const std::string &mac_address_str); diff --git a/libtransport/src/hicn/transport/core/tcp_socket_connector.cc b/libtransport/src/hicn/transport/core/tcp_socket_connector.cc index f1fd4bbac..c82373ae1 100644 --- a/libtransport/src/hicn/transport/core/tcp_socket_connector.cc +++ b/libtransport/src/hicn/transport/core/tcp_socket_connector.cc @@ -21,6 +21,7 @@ #include <hicn/transport/utils/log.h> #include <hicn/transport/utils/object_pool.h> +#include <thread> #include <vector> namespace transport { @@ -78,10 +79,15 @@ void TcpSocketConnector::connect(std::string ip_address, std::string port) { void TcpSocketConnector::send(const uint8_t *packet, std::size_t len, const PacketSentCallback &packet_sent) { - asio::async_write(socket_, asio::buffer(packet, len), - [packet_sent](std::error_code ec, std::size_t /*length*/) { - packet_sent(); - }); + if (packet_sent != 0) { + asio::async_write(socket_, asio::buffer(packet, len), + [packet_sent](std::error_code ec, + std::size_t /*length*/) { packet_sent(); }); + } else { + if (state_ == ConnectorState::CONNECTED) { + asio::write(socket_, asio::buffer(packet, len)); + } + } } void TcpSocketConnector::send(const Packet::MemBufPtr &packet) { @@ -223,38 +229,37 @@ void TcpSocketConnector::tryReconnect() { } void TcpSocketConnector::doConnect() { - asio::async_connect(socket_, endpoint_iterator_, - [this](std::error_code ec, tcp::resolver::iterator) { - if (!ec) { - timer_.cancel(); - state_ = ConnectorState::CONNECTED; - asio::ip::tcp::no_delay noDelayOption(true); - socket_.set_option(noDelayOption); - doReadHeader(); + asio::async_connect( + socket_, endpoint_iterator_, + [this](std::error_code ec, tcp::resolver::iterator) { + if (!ec) { + timer_.cancel(); + state_ = ConnectorState::CONNECTED; + asio::ip::tcp::no_delay noDelayOption(true); + socket_.set_option(noDelayOption); + doReadHeader(); - if (data_available_) { - data_available_ = false; - doWrite(); - } + if (data_available_) { + data_available_ = false; + doWrite(); + } - if (is_reconnection_) { - is_reconnection_ = false; - TRANSPORT_LOGI("Connection recovered!\n"); - on_reconnect_callback_(); - } - } else { - sleep(1); - doConnect(); - } - }); + if (is_reconnection_) { + is_reconnection_ = false; + TRANSPORT_LOGI("Connection recovered!\n"); + on_reconnect_callback_(); + } + } else { + doConnect(); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + }); } bool TcpSocketConnector::checkConnected() { return state_ == ConnectorState::CONNECTED; } -void TcpSocketConnector::enableBurst() { return; } - void TcpSocketConnector::startConnectionTimer() { timer_.expires_from_now(std::chrono::seconds(60)); timer_.async_wait(std::bind(&TcpSocketConnector::handleDeadline, this, diff --git a/libtransport/src/hicn/transport/core/tcp_socket_connector.h b/libtransport/src/hicn/transport/core/tcp_socket_connector.h index 6df2fedff..d755b5e86 100644 --- a/libtransport/src/hicn/transport/core/tcp_socket_connector.h +++ b/libtransport/src/hicn/transport/core/tcp_socket_connector.h @@ -47,8 +47,6 @@ class TcpSocketConnector : public Connector { void close() override; - void enableBurst() override; - void connect(std::string ip_address = "127.0.0.1", std::string port = "9695"); private: diff --git a/libtransport/src/hicn/transport/core/udp_socket_connector.cc b/libtransport/src/hicn/transport/core/udp_socket_connector.cc index 54c0eb978..38945e755 100644 --- a/libtransport/src/hicn/transport/core/udp_socket_connector.cc +++ b/libtransport/src/hicn/transport/core/udp_socket_connector.cc @@ -21,6 +21,7 @@ #include <hicn/transport/utils/log.h> #include <hicn/transport/utils/object_pool.h> +#include <thread> #include <vector> namespace transport { @@ -36,7 +37,6 @@ UdpSocketConnector::UdpSocketConnector( socket_(io_service_), resolver_(io_service_), connection_timer_(io_service_), - connection_timeout_(io_service_), read_msg_(packet_pool_.makePtr(nullptr)), is_reconnection_(false), data_available_(false), @@ -54,10 +54,17 @@ void UdpSocketConnector::connect(std::string ip_address, std::string port) { void UdpSocketConnector::send(const uint8_t *packet, std::size_t len, const PacketSentCallback &packet_sent) { - socket_.async_send(asio::buffer(packet, len), - [packet_sent](std::error_code ec, std::size_t /*length*/) { - packet_sent(); - }); + if (packet_sent != 0) { + socket_.async_send( + asio::buffer(packet, len), + [packet_sent](std::error_code ec, std::size_t /*length*/) { + packet_sent(); + }); + } else { + if (state_ == ConnectorState::CONNECTED) { + socket_.send(asio::buffer(packet, len)); + } + } } void UdpSocketConnector::send(const Packet::MemBufPtr &packet) { @@ -76,6 +83,14 @@ void UdpSocketConnector::send(const Packet::MemBufPtr &packet) { } void UdpSocketConnector::close() { + if (io_service_.stopped()) { + doClose(); + } else { + io_service_.dispatch(std::bind(&UdpSocketConnector::doClose, this)); + } +} + +void UdpSocketConnector::doClose() { if (state_ != ConnectorState::CLOSED) { state_ = ConnectorState::CLOSED; if (socket_.is_open()) { @@ -86,8 +101,6 @@ void UdpSocketConnector::close() { } void UdpSocketConnector::doWrite() { - // TODO improve this piece of code for sending many buffers togethers - // if list contains more than one packet auto packet = output_buffer_.front().get(); auto array = std::vector<asio::const_buffer>(); @@ -97,8 +110,8 @@ void UdpSocketConnector::doWrite() { current = current->next(); } while (current != packet); - socket_.async_send(std::move(array), [this /*, packet*/](std::error_code ec, - std::size_t length) { + socket_.async_send(std::move(array), [this](std::error_code ec, + std::size_t length) { if (TRANSPORT_EXPECT_TRUE(!ec)) { output_buffer_.pop_front(); if (!output_buffer_.empty()) { @@ -139,54 +152,53 @@ void UdpSocketConnector::tryReconnect() { TRANSPORT_LOGE("Connection lost. Trying to reconnect...\n"); state_ = ConnectorState::CONNECTING; is_reconnection_ = true; - connection_timer_.expires_from_now(std::chrono::seconds(1)); - connection_timer_.async_wait([this](const std::error_code &ec) { - if (!ec) { - if (socket_.is_open()) { - socket_.shutdown(asio::ip::tcp::socket::shutdown_type::shutdown_both); - socket_.close(); - } - startConnectionTimer(); - doConnect(); + io_service_.post([this]() { + if (socket_.is_open()) { + socket_.shutdown(asio::ip::tcp::socket::shutdown_type::shutdown_both); + socket_.close(); } + + doConnect(); + startConnectionTimer(); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); }); } } void UdpSocketConnector::doConnect() { - asio::async_connect(socket_, endpoint_iterator_, - [this](std::error_code ec, udp::resolver::iterator) { - if (!ec) { - connection_timeout_.cancel(); - state_ = ConnectorState::CONNECTED; - doRead(); - - if (data_available_) { - data_available_ = false; - doWrite(); - } - - if (is_reconnection_) { - is_reconnection_ = false; - on_reconnect_callback_(); - } - } else { - sleep(1); - doConnect(); - } - }); + asio::async_connect( + socket_, endpoint_iterator_, + [this](std::error_code ec, udp::resolver::iterator) { + if (!ec) { + connection_timer_.cancel(); + state_ = ConnectorState::CONNECTED; + doRead(); + + if (data_available_) { + data_available_ = false; + doWrite(); + } + + if (is_reconnection_) { + is_reconnection_ = false; + } + + on_reconnect_callback_(); + } else { + doConnect(); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + }); } bool UdpSocketConnector::checkConnected() { return state_ == ConnectorState::CONNECTED; } -void UdpSocketConnector::enableBurst() { return; } - void UdpSocketConnector::startConnectionTimer() { - connection_timeout_.expires_from_now(std::chrono::seconds(60)); - connection_timeout_.async_wait(std::bind(&UdpSocketConnector::handleDeadline, - this, std::placeholders::_1)); + connection_timer_.expires_from_now(std::chrono::seconds(60)); + connection_timer_.async_wait(std::bind(&UdpSocketConnector::handleDeadline, + this, std::placeholders::_1)); } void UdpSocketConnector::handleDeadline(const std::error_code &ec) { diff --git a/libtransport/src/hicn/transport/core/udp_socket_connector.h b/libtransport/src/hicn/transport/core/udp_socket_connector.h index 87198efde..7c5dbaf10 100644 --- a/libtransport/src/hicn/transport/core/udp_socket_connector.h +++ b/libtransport/src/hicn/transport/core/udp_socket_connector.h @@ -45,8 +45,6 @@ class UdpSocketConnector : public Connector { void close() override; - void enableBurst() override; - void connect(std::string ip_address = "127.0.0.1", std::string port = "9695"); private: @@ -56,6 +54,8 @@ class UdpSocketConnector : public Connector { void doWrite(); + void doClose(); + bool checkConnected(); private: @@ -70,7 +70,6 @@ class UdpSocketConnector : public Connector { asio::ip::udp::resolver resolver_; asio::ip::udp::resolver::iterator endpoint_iterator_; asio::steady_timer connection_timer_; - asio::steady_timer connection_timeout_; utils::ObjectPool<utils::MemBuf>::Ptr read_msg_; diff --git a/libtransport/src/hicn/transport/http/client_connection.cc b/libtransport/src/hicn/transport/http/client_connection.cc index fb7dbdfac..fadf0ae44 100644 --- a/libtransport/src/hicn/transport/http/client_connection.cc +++ b/libtransport/src/hicn/transport/http/client_connection.cc @@ -26,7 +26,7 @@ namespace http { using namespace transport; HTTPClientConnection::HTTPClientConnection() - : consumer_(TransportProtocolAlgorithms::RAAQM, io_service_), + : consumer_(TransportProtocolAlgorithms::RAAQM), read_bytes_callback_(nullptr), read_buffer_(nullptr), response_(std::make_shared<HTTPResponse>()), diff --git a/libtransport/src/hicn/transport/http/client_connection.h b/libtransport/src/hicn/transport/http/client_connection.h index 6c150f848..d0624702f 100644 --- a/libtransport/src/hicn/transport/http/client_connection.h +++ b/libtransport/src/hicn/transport/http/client_connection.h @@ -88,8 +88,6 @@ class HTTPClientConnection : public ConsumerSocket::ReadCallback { void readError(const std::error_code ec) noexcept override; void readSuccess(std::size_t total_size) noexcept override; - asio::io_service io_service_; - // The consumer socket ConsumerSocket consumer_; diff --git a/libtransport/src/hicn/transport/http/server_acceptor.cc b/libtransport/src/hicn/transport/http/server_acceptor.cc index 486b04c57..e478dfcd4 100644 --- a/libtransport/src/hicn/transport/http/server_acceptor.cc +++ b/libtransport/src/hicn/transport/http/server_acceptor.cc @@ -63,9 +63,7 @@ HTTPServerAcceptor::HTTPServerAcceptor(std::string &server_locator, core::Prefix acceptor_namespace(network, 64); std::string producer_identity = "acceptor_producer"; - acceptor_producer_ = std::make_shared<ProducerSocket>( - io_service_); /*, - utils::Identity::generateIdentity(producer_identity));*/ + acceptor_producer_ = std::make_shared<ProducerSocket>(); acceptor_producer_->registerPrefix(acceptor_namespace); } diff --git a/libtransport/src/hicn/transport/http/server_acceptor.h b/libtransport/src/hicn/transport/http/server_acceptor.h index 4e7350b76..6ed58f70e 100644 --- a/libtransport/src/hicn/transport/http/server_acceptor.h +++ b/libtransport/src/hicn/transport/http/server_acceptor.h @@ -53,7 +53,6 @@ class HTTPServerAcceptor { void processIncomingInterest(ProducerSocket &p, Interest &interest); OnHttpRequest callback_; - asio::io_service io_service_; std::shared_ptr<ProducerSocket> acceptor_producer_; std::map<int, std::shared_ptr<HTTPServerPublisher>> publishers_; diff --git a/libtransport/src/hicn/transport/http/server_publisher.cc b/libtransport/src/hicn/transport/http/server_publisher.cc index 012f36091..6a4bb9c48 100644 --- a/libtransport/src/hicn/transport/http/server_publisher.cc +++ b/libtransport/src/hicn/transport/http/server_publisher.cc @@ -23,7 +23,7 @@ namespace http { HTTPServerPublisher::HTTPServerPublisher(const core::Name &content_name) : content_name_(content_name, true) { std::string identity = "acceptor_producer"; - producer_ = std::make_unique<ProducerSocket>(io_service_); + producer_ = std::make_unique<ProducerSocket>(); // utils::Identity::generateIdentity(identity)); core::Prefix publisher_prefix(content_name_, 128); producer_->registerPrefix(publisher_prefix); diff --git a/libtransport/src/hicn/transport/http/server_publisher.h b/libtransport/src/hicn/transport/http/server_publisher.h index 1f12fd8f9..33d596f63 100644 --- a/libtransport/src/hicn/transport/http/server_publisher.h +++ b/libtransport/src/hicn/transport/http/server_publisher.h @@ -59,7 +59,6 @@ class HTTPServerPublisher { private: Name content_name_; std::unique_ptr<asio::steady_timer> timer_; - asio::io_service io_service_; std::unique_ptr<ProducerSocket> producer_; ProducerInterestCallback interest_enter_callback_; utils::UserCallback wait_callback_; diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc index 41feb4b45..495f8c8f3 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.cc @@ -215,4 +215,4 @@ void RTCProducerSocket::sendNack(const Interest &interest) { } // namespace interface -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h index 6506b506e..be39d2b32 100644 --- a/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/rtc_socket_producer.h @@ -63,4 +63,4 @@ class RTCProducerSocket : public ProducerSocket { } // namespace interface -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index 37d545779..af99fd60c 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -123,10 +123,12 @@ void ConsumerSocket::asyncSendInterest(Interest::Ptr &&interest, } void ConsumerSocket::stop() { - if (transport_protocol_->isRunning()) { - std::cout << "Stopping transport protocol " << std::endl; - transport_protocol_->stop(); - } + auto &io_service = getIoService(); + io_service.dispatch([this]() { + if (transport_protocol_->isRunning()) { + transport_protocol_->stop(); + } + }); } void ConsumerSocket::resume() { @@ -141,4 +143,4 @@ asio::io_service &ConsumerSocket::getIoService() { } // namespace interface -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index 40344af5d..41646c940 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -868,4 +868,4 @@ class ConsumerSocket : public BaseSocket { } // namespace interface -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index c4cf95895..c85b8af32 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -366,4 +366,4 @@ asio::io_service &ProducerSocket::getIoService() { return io_service_; } } // namespace interface -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.h b/libtransport/src/hicn/transport/interfaces/socket_producer.h index 200c32a95..744ddd86d 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.h @@ -598,4 +598,4 @@ class ProducerSocket : public Socket<BasePortal>, } // namespace interface -} // namespace transport +} // namespace transport
\ No newline at end of file diff --git a/libtransport/src/hicn/transport/protocols/rtc.cc b/libtransport/src/hicn/transport/protocols/rtc.cc index c9aa6a56a..b514d0587 100644 --- a/libtransport/src/hicn/transport/protocols/rtc.cc +++ b/libtransport/src/hicn/transport/protocols/rtc.cc @@ -739,4 +739,4 @@ void RTCTransportProtocol::returnContentToApplication( } // end namespace protocol -} // end namespace transport +} // end namespace transport
\ No newline at end of file diff --git a/utils/src/hiperf.cc b/utils/src/hiperf.cc index ddc79d520..681696dbf 100644 --- a/utils/src/hiperf.cc +++ b/utils/src/hiperf.cc @@ -467,6 +467,7 @@ class HIperfClient { void readError(const std::error_code ec) noexcept override { std::cerr << "Error while reading from RTC socket" << std::endl; + client_.io_service_.stop(); } void readSuccess(std::size_t total_size) noexcept override { @@ -509,6 +510,7 @@ class HIperfClient { void readError(const std::error_code ec) noexcept override { std::cerr << "Error " << ec.message() << " while reading from socket" << std::endl; + client_.io_service_.stop(); } void readSuccess(std::size_t total_size) noexcept override { |