From 51ff9b669dff18c9300b9fe5bdef91e7040edac0 Mon Sep 17 00:00:00 2001 From: Mauro Sardara Date: Tue, 29 Jan 2019 15:49:15 +0100 Subject: [HICN-17] Add possibility to destroy connection directly from stopEventsLoop with an additional parameter. Change-Id: I869a079a7b2f436768a62de66fd9281a7d1243cd Signed-off-by: Mauro Sardara --- .../src/hicn/transport/core/memif_connector.cc | 5 +++-- libtransport/src/hicn/transport/core/portal.h | 16 +++++++--------- .../src/hicn/transport/core/socket_connector.cc | 19 +++++++++++++++++-- .../src/hicn/transport/core/socket_connector.h | 1 + .../src/hicn/transport/interfaces/socket_consumer.cc | 4 ++-- .../src/hicn/transport/interfaces/socket_consumer.h | 3 ++- .../src/hicn/transport/interfaces/socket_producer.cc | 6 ++---- 7 files changed, 34 insertions(+), 20 deletions(-) (limited to 'libtransport/src') diff --git a/libtransport/src/hicn/transport/core/memif_connector.cc b/libtransport/src/hicn/transport/core/memif_connector.cc index b3785e5c3..6c5f2ff5f 100644 --- a/libtransport/src/hicn/transport/core/memif_connector.cc +++ b/libtransport/src/hicn/transport/core/memif_connector.cc @@ -38,7 +38,6 @@ MemifConnector::MemifConnector(PacketReceivedCallback &&receive_callback, timer_set_(false), send_timer_(std::make_unique(event_reactor_)), io_service_(io_service), - work_(std::make_unique(io_service_)), packet_counter_(0), memif_connection_({}), tx_buf_counter_(0), @@ -74,6 +73,8 @@ void MemifConnector::connect(uint32_t memif_id, long memif_mode) { createMemif(memif_id, memif_mode, nullptr); + work_ = std::make_unique(io_service_); + while (is_connecting_) { MemifConnector::main_event_reactor_.runOneEvent(); } @@ -402,7 +403,7 @@ void MemifConnector::close() { if (!closed_) { closed_ = true; event_reactor_.stop(); - io_service_.stop(); + work_.reset(); if (memif_worker_ && memif_worker_->joinable()) { memif_worker_->join(); diff --git a/libtransport/src/hicn/transport/core/portal.h b/libtransport/src/hicn/transport/core/portal.h index 52a721a35..58406bbde 100644 --- a/libtransport/src/hicn/transport/core/portal.h +++ b/libtransport/src/hicn/transport/core/portal.h @@ -97,9 +97,7 @@ class Portal { virtual void onInterest(Interest::Ptr &&i) = 0; }; - Portal() : Portal(internal_io_service_) { - internal_work_ = std::make_unique(io_service_); - } + Portal() : Portal(internal_io_service_) {} Portal(asio::io_service &io_service) : io_service_(io_service), @@ -130,8 +128,7 @@ class Portal { } ~Portal() { - connector_.close(); - stopEventsLoop(); + stopEventsLoop(true); } TRANSPORT_ALWAYS_INLINE bool interestIsPending(const Name &name) { @@ -227,15 +224,17 @@ class Portal { forwarder_interface_.send(content_object); } - TRANSPORT_ALWAYS_INLINE void stopEventsLoop() { - internal_work_.reset(); - + TRANSPORT_ALWAYS_INLINE void stopEventsLoop(bool kill_connection = false) { for (auto &pend_interest : pending_interest_hash_table_) { pend_interest.second->cancelTimer(); } clear(); + if(kill_connection) { + connector_.close(); + } + io_service_.post([this]() { io_service_.stop(); }); } @@ -340,7 +339,6 @@ class Portal { private: asio::io_service &io_service_; asio::io_service internal_io_service_; - std::unique_ptr internal_work_; std::string app_name_; diff --git a/libtransport/src/hicn/transport/core/socket_connector.cc b/libtransport/src/hicn/transport/core/socket_connector.cc index 332b87ec7..704d3badb 100644 --- a/libtransport/src/hicn/transport/core/socket_connector.cc +++ b/libtransport/src/hicn/transport/core/socket_connector.cc @@ -62,6 +62,7 @@ SocketConnector::SocketConnector(PacketReceivedCallback &&receive_callback, is_connecting_(false), is_reconnection_(false), data_available_(false), + is_closed_(false), receive_callback_(receive_callback), on_reconnect_callback_(on_reconnect_callback), app_name_(app_name) {} @@ -102,7 +103,11 @@ void SocketConnector::send(const Packet::MemBufPtr &packet) { } void SocketConnector::close() { - io_service_.post([this]() { socket_.close(); }); + io_service_.dispatch([this]() { + is_closed_ = true; + socket_.shutdown(asio::ip::tcp::socket::shutdown_type::shutdown_both); + socket_.close(); + }); } void SocketConnector::doWrite() { @@ -125,6 +130,9 @@ void SocketConnector::doWrite() { if (!output_buffer_.empty()) { doWrite(); } + } else if (ec.value() == static_cast(std::errc::operation_canceled)) { + // The connection has been closed by the application. + return; } else { TRANSPORT_LOGE("%d %s", ec.value(), ec.message().c_str()); tryReconnect(); @@ -141,6 +149,9 @@ void SocketConnector::doReadBody(std::size_t body_length) { if (TRANSPORT_EXPECT_TRUE(!ec)) { receive_callback_(std::move(read_msg_)); doReadHeader(); + } else if (ec.value() == static_cast(std::errc::operation_canceled)) { + // The connection has been closed by the application. + return; } else { TRANSPORT_LOGE("%d %s", ec.value(), ec.message().c_str()); tryReconnect(); @@ -165,6 +176,9 @@ void SocketConnector::doReadHeader() { } else { TRANSPORT_LOGE("Decoding error. Ignoring packet."); } + } else if (ec.value() == static_cast(std::errc::operation_canceled)) { + // The connection has been closed by the application. + return; } else { TRANSPORT_LOGE("%d %s", ec.value(), ec.message().c_str()); tryReconnect(); @@ -173,11 +187,12 @@ void SocketConnector::doReadHeader() { } void SocketConnector::tryReconnect() { - if (!is_connecting_) { + if (!is_connecting_ && !is_closed_) { TRANSPORT_LOGE("Connection lost. Trying to reconnect...\n"); is_connecting_ = true; is_reconnection_ = true; io_service_.post([this]() { + socket_.shutdown(asio::ip::tcp::socket::shutdown_type::shutdown_both); socket_.close(); startConnectionTimer(); doConnect(); diff --git a/libtransport/src/hicn/transport/core/socket_connector.h b/libtransport/src/hicn/transport/core/socket_connector.h index b1757e59a..e014111e2 100644 --- a/libtransport/src/hicn/transport/core/socket_connector.h +++ b/libtransport/src/hicn/transport/core/socket_connector.h @@ -79,6 +79,7 @@ class SocketConnector : public Connector { bool is_connecting_; bool is_reconnection_; bool data_available_; + bool is_closed_; PacketReceivedCallback receive_callback_; OnReconnect on_reconnect_callback_; diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc index 27ed4e65f..89411e92c 100644 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.cc @@ -99,7 +99,7 @@ int ConsumerSocket::consume(const Name &name, transport_protocol_->start(receive_buffer); - return CONSUMER_READY; + return CONSUMER_FINISHED; } int ConsumerSocket::asyncConsume( @@ -115,7 +115,7 @@ int ConsumerSocket::asyncConsume( }); } - return CONSUMER_READY; + return CONSUMER_RUNNING; } void ConsumerSocket::asyncSendInterest(Interest::Ptr &&interest, diff --git a/libtransport/src/hicn/transport/interfaces/socket_consumer.h b/libtransport/src/hicn/transport/interfaces/socket_consumer.h index 9e309aae8..536d2fde3 100755 --- a/libtransport/src/hicn/transport/interfaces/socket_consumer.h +++ b/libtransport/src/hicn/transport/interfaces/socket_consumer.h @@ -26,8 +26,9 @@ #include #include -#define CONSUMER_READY 0 +#define CONSUMER_FINISHED 0 #define CONSUMER_BUSY 1 +#define CONSUMER_RUNNING 2 namespace transport { diff --git a/libtransport/src/hicn/transport/interfaces/socket_producer.cc b/libtransport/src/hicn/transport/interfaces/socket_producer.cc index 69adc2b3f..d9204f111 100755 --- a/libtransport/src/hicn/transport/interfaces/socket_producer.cc +++ b/libtransport/src/hicn/transport/interfaces/socket_producer.cc @@ -54,9 +54,9 @@ ProducerSocket::ProducerSocket(asio::io_service &io_service) } ProducerSocket::~ProducerSocket() { - TRANSPORT_LOGI("Destroying the ProducerSocket"); + processing_thread_stop_ = true; - portal_->stopEventsLoop(); + portal_->stopEventsLoop(true); if (processing_thread_.joinable()) { processing_thread_.join(); @@ -79,8 +79,6 @@ void ProducerSocket::serveForever() { } void ProducerSocket::stop() { - TRANSPORT_LOGI("Calling stop for ProducerSocket"); - portal_->killConnection(); portal_->stopEventsLoop(); } -- cgit 1.2.3-korg