3.5.6 new commits

implementation/endpoints/include/local_tcp_client_endpoint_impl.hpp

@@ -67,6 +67,9 @@ class local_tcp_client_endpoint_impl: public local_tcp_client_endpoint_base_impl
 
     // send data
     message_buffer_ptr_t send_data_buffer_;
+
+    bool is_stopping_;
+    std::condition_variable_any queue_cv_;
 };
 
 } // namespace vsomeip_v3

implementation/endpoints/src/client_endpoint_impl.cpp

@@ -477,7 +477,7 @@ void client_endpoint_impl<Protocol>::cancel_and_connect_cbk(
         std::lock_guard<std::mutex> its_lock(connecting_timer_mutex_);
         connecting_timer_state_ = connecting_timer_state_e::FINISH_ERROR;
         operations_cancelled = connecting_timer_.cancel();
-        if (operations_cancelled != 0) {
+        if (!_error) {
             connecting_timer_state_ = connecting_timer_state_e::FINISH_SUCCESS;
         }
         connecting_timer_condition_.notify_all();

implementation/endpoints/src/local_tcp_client_endpoint_impl.cpp

@@ -51,6 +51,7 @@ void local_tcp_client_endpoint_impl::restart(bool _force) {
         sending_blocked_ = false;
         queue_.clear();
         queue_size_ = 0;
+        is_sending_ = false;
     }
     {
         std::lock_guard<std::mutex> its_lock(socket_mutex_);
@@ -71,6 +72,7 @@ void local_tcp_client_endpoint_impl::start() {
         {
             std::lock_guard<std::recursive_mutex> its_lock(mutex_);
             sending_blocked_ = false;
+            is_stopping_ = false;
         }
         connect();
     }
@@ -80,6 +82,7 @@ void local_tcp_client_endpoint_impl::stop() {
     {
         std::lock_guard<std::recursive_mutex> its_lock(mutex_);
         sending_blocked_ = true;
+        is_stopping_ = true;
     }
     {
         std::lock_guard<std::mutex> its_lock(connect_timer_mutex_);
@@ -98,13 +101,13 @@ void local_tcp_client_endpoint_impl::stop() {
         std::uint32_t times_slept(0);
 
         while (times_slept <= LOCAL_TCP_WAIT_SEND_QUEUE_ON_STOP) {
-            mutex_.lock();
+            std::unique_lock<std::recursive_mutex> its_lock(mutex_);
             send_queue_empty = (queue_.size() == 0);
-            mutex_.unlock();
             if (send_queue_empty) {
                 break;
             } else {
-                std::this_thread::sleep_for(std::chrono::milliseconds(10));
+                queue_cv_.wait_for(its_lock, std::chrono::milliseconds(10),
+                                   [this] { return queue_.size() == 0; });
                 times_slept++;
             }
         }
@@ -277,6 +280,7 @@ void local_tcp_client_endpoint_impl::send_queued(std::pair<message_buffer_ptr_t,
         } else {
             VSOMEIP_WARNING << "ltcei::" << __func__ << ": try to send while socket was not open | endpoint > " << this;
             was_not_connected_ = true;
+            is_sending_ = false;
         }
     }
 }
@@ -311,6 +315,10 @@ void local_tcp_client_endpoint_impl::receive_cbk(
             sending_blocked_ = false;
             queue_.clear();
             queue_size_ = 0;
+
+            if (is_stopping_) {
+                queue_cv_.notify_all();
+            }
         }
         error_handler_t handler;
         {

implementation/endpoints/src/local_uds_client_endpoint_impl.cpp

@@ -54,6 +54,7 @@ void local_uds_client_endpoint_impl::restart(bool _force) {
         sending_blocked_ = false;
         queue_.clear();
         queue_size_ = 0;
+        is_sending_ = false;
     }
     {
         std::lock_guard<std::mutex> its_lock(socket_mutex_);
@@ -233,18 +234,21 @@ void local_uds_client_endpoint_impl::send_queued(std::pair<message_buffer_ptr_t,
 
     {
         std::lock_guard<std::mutex> its_lock(socket_mutex_);
+
+        auto its_me {std::dynamic_pointer_cast<local_uds_client_endpoint_impl>(shared_from_this())};
+        auto buffer_ptr = _entry.first; // Capture shared_ptr to ensure it stays alive
+
         if(socket_->is_open()) {
             boost::asio::async_write(
-                *socket_, bufs,
-                strand_.wrap(std::bind(&client_endpoint_impl::send_cbk,
-                    std::dynamic_pointer_cast<local_uds_client_endpoint_impl>(
-                        shared_from_this()),
-                        std::placeholders::_1, std::placeholders::_2,
-                        _entry.first)));
+                    *socket_, bufs,
+                    strand_.wrap([its_me, buffer_ptr](const boost::system::error_code& ec, std::size_t bytes_transferred){
+                        its_me->send_cbk(ec, bytes_transferred, buffer_ptr);
+                    }));
         } else {
             VSOMEIP_WARNING << "lucei::" << __func__ << ": try to send while socket was not open | endpoint > " << this;
             was_not_connected_ = true;
-        }
+            is_sending_ = false;
+        } 
     }
 }
 

implementation/endpoints/src/tcp_client_endpoint_impl.cpp

@@ -115,6 +115,7 @@ void tcp_client_endpoint_impl::restart(bool _force) {
             }
             self->queue_.clear();
             self->queue_size_ = 0;
+            self->is_sending_ = false;
         }
         VSOMEIP_WARNING << "tce::restart: local: " << address_port_local
                 << " remote: " << self->get_address_port_remote();
@@ -380,6 +381,7 @@ void tcp_client_endpoint_impl::send_queued(std::pair<message_buffer_ptr_t, uint3
         } else {
             VSOMEIP_WARNING << "tcei::" << __func__ << ": try to send while socket was not open | endpoint > " << this;
             was_not_connected_ = true;
+            is_sending_ = false;
         }
     }
 }

implementation/endpoints/src/tcp_server_endpoint_impl.cpp

@@ -260,10 +260,12 @@ void tcp_server_endpoint_impl::accept_cbk(connection::ptr _connection,
                 VSOMEIP_WARNING << "tsei::" << __func__
                                 << ": couldn't enable keep_alive: " << its_error.message();
             }
-            // Setting the TIME_WAIT to 0 seconds forces RST to always be sent in reponse to a FIN
-            // The linger is needed to be set for suspend to RAM, since after resuming the
-            // connection could be lost and the socket could get stuck in TIME_WAIT for 120 seconds
-            new_connection_socket.set_option(boost::asio::socket_base::linger(true, 5), its_error);
+
+            // force always TCP RST on close/shutdown, in order to:
+            // 1) avoid issues with TIME_WAIT, which otherwise lasts for 120 secs with a
+            // non-responding endpoint (see also 4396812d2)
+            // 2) handle by default what needs to happen at suspend/shutdown
+            new_connection_socket.set_option(boost::asio::socket_base::linger(true, 0), its_error);
             if (its_error) {
                 VSOMEIP_WARNING << "tsei::" << __func__
                                 << ": setting SO_LINGER failed: " << its_error.message();
@@ -424,17 +426,6 @@ void tcp_server_endpoint_impl::connection::stop() {
     if (socket_.is_open()) {
         boost::system::error_code its_error;
 
-        auto its_server {server_.lock()};
-
-        if (its_server && its_server->is_suspended()) {
-            socket_.set_option(boost::asio::socket_base::linger(true, 0), its_error);
-            if (its_error) {
-                VSOMEIP_WARNING << "tcp_server_endpoint_impl::connection::stop<"
-                                << get_address_port_remote() << ">:setting SO_LINGER failed ("
-                                << its_error.message() << ")";
-            }
-        }
-
         socket_.shutdown(socket_.shutdown_both, its_error);
         if (its_error) {
             VSOMEIP_WARNING << "tcp_server_endpoint_impl::connection::stop<"

implementation/endpoints/src/udp_client_endpoint_impl.cpp

@@ -205,6 +205,7 @@ void udp_client_endpoint_impl::restart(bool _force) {
         local = get_address_port_local();
     }
     was_not_connected_ = true;
+    is_sending_ = false;
     reconnect_counter_ = 0;
     VSOMEIP_WARNING << "uce::restart: local: " << local
             << " remote: " << get_address_port_remote();

implementation/endpoints/src/udp_server_endpoint_impl.cpp

@@ -328,23 +328,31 @@ void udp_server_endpoint_impl::receive() {
 }
 
 void udp_server_endpoint_impl::receive_unicast() {
-
     std::lock_guard<std::mutex> its_lock(unicast_mutex_);
-
-    if (!is_stopped_ && unicast_socket_->is_open()) {
+    if (!is_stopped_ && unicast_socket_ && unicast_socket_->is_open()) {
         unicast_socket_->async_receive_from(
                 boost::asio::buffer(&unicast_recv_buffer_[0], max_message_size_), unicast_remote_,
                 std::bind(&udp_server_endpoint_impl::on_unicast_received,
                           std::dynamic_pointer_cast<udp_server_endpoint_impl>(shared_from_this()),
                           std::placeholders::_1, std::placeholders::_2));
+    } else {
+        if (is_stopped_ && unicast_socket_) {
+            io_.post(std::bind(
+                    &udp_server_endpoint_impl::shutdown_and_close,
+                    std::dynamic_pointer_cast<udp_server_endpoint_impl>(shared_from_this()), true));
+        } else {
+            VSOMEIP_WARNING << "usei::" << __func__
+                            << ": Endpoint is running, but the unicast socket is not existing/"
+                               "closed. (0x"
+                            << std::hex << this << ")";
+        }
     }
 }
 
 //
 // receive_multicast is called with multicast_mutex_ being hold
 //
 void udp_server_endpoint_impl::receive_multicast(uint8_t _multicast_id) {
-
     if (!is_stopped_ && _multicast_id == multicast_id_ && multicast_socket_
         && multicast_socket_->is_open()) {
         auto its_storage = std::make_shared<udp_endpoint_receive_op::storage>(
@@ -357,6 +365,15 @@ void udp_server_endpoint_impl::receive_multicast(uint8_t _multicast_id) {
                 boost::asio::ip::address(), std::numeric_limits<std::size_t>::min());
         multicast_socket_->async_wait(socket_type::wait_read,
                                       udp_endpoint_receive_op::receive_cb(its_storage));
+    } else {
+        VSOMEIP_WARNING << "usei::" << __func__ << ": is_stopped_: " << is_stopped_ << std::dec
+                        << " _multicast_id: " << static_cast<int>(_multicast_id)
+                        << " multicast_id_: " << static_cast<int>(multicast_id_);
+        if (is_stopped_ && multicast_socket_) {
+            io_.post(std::bind(
+                    &udp_server_endpoint_impl::shutdown_and_close,
+                    std::dynamic_pointer_cast<udp_server_endpoint_impl>(shared_from_this()), false));
+        }
     }
 }
 
@@ -1029,7 +1046,7 @@ void udp_server_endpoint_impl::set_multicast_option(const boost::asio::ip::addre
 
             if (0 == joined_.size()) {
                 multicast_socket_->cancel(_error);
-
+                multicast_id_ = 0;
                 multicast_socket_.reset();
                 multicast_local_.reset(nullptr);
             }

implementation/routing/src/routing_manager_base.cpp

@@ -773,7 +773,11 @@ void routing_manager_base::add_known_client(client_t _client, const std::string
     }
 #endif
     std::lock_guard<std::mutex> its_lock(known_clients_mutex_);
-    known_clients_[_client] = _client_host;
+    if (known_clients_.find(_client) == known_clients_.end()) {
+        known_clients_[_client] = _client_host;
+    } else if (!_client_host.empty()) {
+        known_clients_[_client] = _client_host;
+    }
 }
 
 void routing_manager_base::remove_known_client(client_t _client) {
@@ -1188,20 +1192,13 @@ void routing_manager_base::remove_local(client_t _client,
     if (_remove_sec_client) {
         configuration_->get_policy_manager()->remove_client_to_sec_client_mapping(_client);
     }
-
-    auto self {shared_from_this()};
     for (auto its_subscription : _subscribed_eventgroups) {
-        auto its_service = std::get<0>(its_subscription);
-        auto its_instance = std::get<1>(its_subscription);
-        auto its_eventgroup = std::get<2>(its_subscription);
-        host_->on_subscription(its_service, its_instance, its_eventgroup, _client, &its_sec_client,
-                               get_env(_client), false,
-                               [self, this, _client, its_sec_client, its_service, its_instance,
-                                its_eventgroup](const bool _subscription_accepted) {
-                                   (void)_subscription_accepted;
-                                   unsubscribe(_client, &its_sec_client, its_service, its_instance,
-                                               its_eventgroup, ANY_EVENT);
-                               });
+        host_->on_subscription(std::get<0>(its_subscription), std::get<1>(its_subscription),
+                std::get<2>(its_subscription), _client,
+                &its_sec_client, get_env(_client),
+                false, [](const bool _subscription_accepted){ (void)_subscription_accepted; });
+        routing_manager_base::unsubscribe(_client, &its_sec_client, std::get<0>(its_subscription),
+                std::get<1>(its_subscription), std::get<2>(its_subscription), ANY_EVENT);
     }
     ep_mgr_->remove_local(_client);
     remove_known_client(_client);