diff --git a/ecal/core/src/pubsub/ecal_publisher_impl.cpp b/ecal/core/src/pubsub/ecal_publisher_impl.cpp index 437217e9d8..13f61d4429 100644 --- a/ecal/core/src/pubsub/ecal_publisher_impl.cpp +++ b/ecal/core/src/pubsub/ecal_publisher_impl.cpp @@ -104,6 +104,64 @@ namespace namespace eCAL { + void CPublisherImpl::SSendLayerConnectionCounters::Increment(TransportLayer::eType layer_) + { + switch (layer_) + { + case TransportLayer::eType::udp_mc: + udp.fetch_add(1, std::memory_order_relaxed); + break; + case TransportLayer::eType::shm: + shm.fetch_add(1, std::memory_order_relaxed); + break; + case TransportLayer::eType::tcp: + tcp.fetch_add(1, std::memory_order_relaxed); + break; + default: + break; + } + } + + void CPublisherImpl::SSendLayerConnectionCounters::Decrement(TransportLayer::eType layer_) + { + switch (layer_) + { + case TransportLayer::eType::udp_mc: + udp.fetch_sub(1, std::memory_order_relaxed); + break; + case TransportLayer::eType::shm: + shm.fetch_sub(1, std::memory_order_relaxed); + break; + case TransportLayer::eType::tcp: + tcp.fetch_sub(1, std::memory_order_relaxed); + break; + default: + break; + } + } + + void CPublisherImpl::SSendLayerConnectionCounters::Reset() + { + udp.store(0, std::memory_order_relaxed); + shm.store(0, std::memory_order_relaxed); + tcp.store(0, std::memory_order_relaxed); + } + + bool CPublisherImpl::SSendLayerConnectionCounters::UdpEnabled() const + { + return (udp.load(std::memory_order_relaxed) > 0); + } + + bool CPublisherImpl::SSendLayerConnectionCounters::ShmEnabled() const + { + return (shm.load(std::memory_order_relaxed) > 0); + } + + bool CPublisherImpl::SSendLayerConnectionCounters::TcpEnabled() const + { + return (tcp.load(std::memory_order_relaxed) > 0); + } + CPublisherImpl::CPublisherImpl(const SDataTypeInformation& topic_info_, const eCAL::eCALWriter::SAttributes& attr_, SPublisherGlobalContext global_context_) : m_publisher_id(eCAL::Util::GenerateUniqueEntityId()) , m_topic_info(topic_info_) @@ -141,6 +199,7 @@ namespace eCAL { const std::lock_guard lock(m_connection_map_mutex); m_connection_map.clear(); + m_connection_count.store(0, std::memory_order_relaxed); } // mark as no more created @@ -155,6 +214,10 @@ namespace eCAL // get payload buffer size (one time, to avoid multiple computations) const size_t payload_buf_size(payload_.GetSize()); + const bool udp_send_enabled = m_send_layer_connection_counters.UdpEnabled(); + const bool shm_send_enabled = m_send_layer_connection_counters.ShmEnabled(); + const bool tcp_send_enabled = m_send_layer_connection_counters.TcpEnabled(); + // are we allowed to perform zero copy writing? bool allow_zero_copy(false); #if ECAL_CORE_TRANSPORT_SHM @@ -162,11 +225,11 @@ namespace eCAL #endif #if ECAL_CORE_TRANSPORT_UDP // udp is active -> no zero copy - allow_zero_copy &= !m_writer_udp; + allow_zero_copy &= !(m_writer_udp && udp_send_enabled); #endif #if ECAL_CORE_TRANSPORT_TCP // tcp is active -> no zero copy - allow_zero_copy &= !m_writer_tcp; + allow_zero_copy &= !(m_writer_tcp && tcp_send_enabled); #endif // create a payload copy for all layer @@ -186,7 +249,7 @@ namespace eCAL // SHM //////////////////////////////////////////////////////////////////////////// #if ECAL_CORE_TRANSPORT_SHM - if (m_writer_shm) + if (m_writer_shm && shm_send_enabled) { #ifndef NDEBUG eCAL::Logging::Log(Logging::log_level_debug3, m_attributes.topic_name + "::CPublisherImpl::Write::SHM"); @@ -249,7 +312,7 @@ namespace eCAL // UDP (MC) //////////////////////////////////////////////////////////////////////////// #if ECAL_CORE_TRANSPORT_UDP - if (m_writer_udp) + if (m_writer_udp && udp_send_enabled) { #ifndef NDEBUG eCAL::Logging::Log(Logging::log_level_debug3, m_attributes.topic_name + "::CPublisherImpl::Write::udp"); @@ -298,7 +361,7 @@ namespace eCAL // TCP //////////////////////////////////////////////////////////////////////////// #if ECAL_CORE_TRANSPORT_TCP - if (m_writer_tcp) + if (m_writer_tcp && tcp_send_enabled) { #ifndef NDEBUG eCAL::Logging::Log(Logging::log_level_debug3, m_attributes.topic_name + "::CPublisherImpl::Send::TCP"); @@ -406,8 +469,8 @@ namespace eCAL #endif // determine if we need to start a transport layer - const TransportLayer::eType layer2activate = DetermineTransportLayer2Start(pub_layers, sub_layers, m_attributes.host_name == subscription_info_.host_name); - switch (layer2activate) + const TransportLayer::eType transport_layer_for_subscription = DetermineTransportLayer(pub_layers, sub_layers, m_attributes.host_name == subscription_info_.host_name); + switch (transport_layer_for_subscription) { case TransportLayer::eType::udp_mc: StartUdpLayer(); @@ -446,26 +509,40 @@ namespace eCAL if (subscription_info_iter == m_connection_map.end()) { // add subscriber to connection map, connection state false - m_connection_map[subscription_info_] = SConnection{ data_type_info_, sub_layer_states_, false }; + m_connection_map[subscription_info_] = SConnection{ data_type_info_, sub_layer_states_, transport_layer_for_subscription, false }; } else { // existing connection, we got the second update now auto& connection = subscription_info_iter->second; + const bool was_active = connection.state; // if this connection was inactive before // activate it now and flag a new connection finally - if (!connection.state) + if (!was_active) { is_new_connection = true; + m_connection_count.fetch_add(1, std::memory_order_relaxed); + // first active update: selected layer becomes effective for sending + connection = SConnection{ data_type_info_, sub_layer_states_, transport_layer_for_subscription, true }; + m_send_layer_connection_counters.Increment(connection.selected_layer); } + else + { + // selected_layer is expected to stay stable for active eCAL connections +#ifndef NDEBUG + if (connection.selected_layer != transport_layer_for_subscription) + { + eCAL::Logging::Log(Logging::log_level_warning, m_attributes.topic_name + "::CPublisherImpl::ApplySubscriberRegistration - selected transport layer changed unexpectedly for active connection"); + } +#endif - // update the data type, the layer states and set the state active - connection = SConnection{ data_type_info_, sub_layer_states_, true }; + // keep selected_layer unchanged and refresh metadata/state + connection.data_type_info = data_type_info_; + connection.layer_states = sub_layer_states_; + connection.state = true; + } } - - // update connection count - m_connection_count = GetConnectionCount(); } @@ -497,11 +574,18 @@ namespace eCAL { const std::lock_guard lock(m_connection_map_mutex); - // remove key from connection map - m_connection_map.erase(subscription_info_); + auto subscription_info_iter = m_connection_map.find(subscription_info_); + if (subscription_info_iter != m_connection_map.end()) + { + if (subscription_info_iter->second.state) + { + m_connection_count.fetch_sub(1, std::memory_order_relaxed); + m_send_layer_connection_counters.Decrement(subscription_info_iter->second.selected_layer); + } - // update connection count - m_connection_count = GetConnectionCount(); + // remove key from connection map + m_connection_map.erase(subscription_info_iter); + } } // fire disconnect event @@ -702,20 +786,6 @@ namespace eCAL FireEvent(ePublisherEvent::disconnected, subscription_info_, data_type_info_); } - size_t CPublisherImpl::GetConnectionCount() - { - // no need to lock map here for now, map locked by caller - size_t count(0); - for (const auto& sub : m_connection_map) - { - if (sub.second.state) - { - count++; - } - } - return count; - } - bool CPublisherImpl::StartUdpLayer() { #if ECAL_CORE_TRANSPORT_UDP @@ -819,6 +889,8 @@ namespace eCAL // destroy writer m_writer_tcp.reset(); #endif + + m_send_layer_connection_counters.Reset(); } size_t CPublisherImpl::PrepareWrite(long long id_, size_t len_) @@ -840,7 +912,7 @@ namespace eCAL return snd_hash; } - TransportLayer::eType CPublisherImpl::DetermineTransportLayer2Start(const std::vector& enabled_pub_layer_, const std::vector& enabled_sub_layer_, bool same_host_) + TransportLayer::eType CPublisherImpl::DetermineTransportLayer(const std::vector& enabled_pub_layer_, const std::vector& enabled_sub_layer_, bool same_host_) { // determine the priority list to use const Publisher::Configuration::LayerPriorityVector& layer_priority_vector = same_host_ ? m_attributes.layer_priority_local : m_attributes.layer_priority_remote; diff --git a/ecal/core/src/pubsub/ecal_publisher_impl.h b/ecal/core/src/pubsub/ecal_publisher_impl.h index 1f0a37216b..2998688ef3 100644 --- a/ecal/core/src/pubsub/ecal_publisher_impl.h +++ b/ecal/core/src/pubsub/ecal_publisher_impl.h @@ -130,11 +130,9 @@ namespace eCAL void FireConnectEvent (const SSubscriptionInfo& subscription_info_, const SDataTypeInformation& data_type_info_); void FireDisconnectEvent(const SSubscriptionInfo& subscription_info_, const SDataTypeInformation& data_type_info_); - size_t GetConnectionCount(); - size_t PrepareWrite(long long id_, size_t len_); - TransportLayer::eType DetermineTransportLayer2Start(const std::vector& enabled_pub_layer_, const std::vector& enabled_sub_layer_, bool same_host_); + TransportLayer::eType DetermineTransportLayer(const std::vector& enabled_pub_layer_, const std::vector& enabled_sub_layer_, bool same_host_); int32_t GetFrequency(); @@ -150,12 +148,30 @@ namespace eCAL { SDataTypeInformation data_type_info; SLayerStates layer_states; + TransportLayer::eType selected_layer = TransportLayer::eType::none; bool state = false; }; using SSubscriptionMapT = std::map; + + struct SSendLayerConnectionCounters + { + void Increment(TransportLayer::eType layer_); + void Decrement(TransportLayer::eType layer_); + void Reset(); + + bool UdpEnabled() const; + bool ShmEnabled() const; + bool TcpEnabled() const; + + std::atomic udp{ 0 }; + std::atomic shm{ 0 }; + std::atomic tcp{ 0 }; + }; + mutable std::mutex m_connection_map_mutex; SSubscriptionMapT m_connection_map; std::atomic m_connection_count{ 0 }; + SSendLayerConnectionCounters m_send_layer_connection_counters; std::mutex m_event_id_callback_mutex; PubEventCallbackT m_event_id_callback;