Skip to content

[core] Gate publisher sends by active connection layer #2680

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
KerstinKeller wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[core] Gate publisher sends by active connection layer #2680

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bce1e9d
[core] Gate publisher sends by active connection layer
KerstinKeller Jul 9, 2026
2f26a1c
Store connection counts instead of booleans to better track state.
KerstinKeller Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
138 changes: 105 additions & 33 deletions ecal/core/src/pubsub/ecal_publisher_impl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_)
Expand Down Expand Up @@ -141,6 +199,7 @@ namespace eCAL
{
const std::lock_guard<std::mutex> lock(m_connection_map_mutex);
m_connection_map.clear();
m_connection_count.store(0, std::memory_order_relaxed);
}

// mark as no more created
Expand All @@ -155,18 +214,22 @@ 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
allow_zero_copy = m_attributes.shm.zero_copy_mode; // zero copy mode activated by user
#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
Expand All @@ -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");
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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();
}


Expand Down Expand Up @@ -497,11 +574,18 @@ namespace eCAL
{
const std::lock_guard<std::mutex> 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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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_)
Expand All @@ -840,7 +912,7 @@ namespace eCAL
return snd_hash;
}

TransportLayer::eType CPublisherImpl::DetermineTransportLayer2Start(const std::vector<eTLayerType>& enabled_pub_layer_, const std::vector<eTLayerType>& enabled_sub_layer_, bool same_host_)
TransportLayer::eType CPublisherImpl::DetermineTransportLayer(const std::vector<eTLayerType>& enabled_pub_layer_, const std::vector<eTLayerType>& 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;
Expand Down
22 changes: 19 additions & 3 deletions ecal/core/src/pubsub/ecal_publisher_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<eTLayerType>& enabled_pub_layer_, const std::vector<eTLayerType>& enabled_sub_layer_, bool same_host_);
TransportLayer::eType DetermineTransportLayer(const std::vector<eTLayerType>& enabled_pub_layer_, const std::vector<eTLayerType>& enabled_sub_layer_, bool same_host_);

int32_t GetFrequency();

Expand All @@ -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<SSubscriptionInfo, SConnection>;

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<size_t> udp{ 0 };
std::atomic<size_t> shm{ 0 };
std::atomic<size_t> tcp{ 0 };
};

mutable std::mutex m_connection_map_mutex;
SSubscriptionMapT m_connection_map;
std::atomic<size_t> m_connection_count{ 0 };
SSendLayerConnectionCounters m_send_layer_connection_counters;

std::mutex m_event_id_callback_mutex;
PubEventCallbackT m_event_id_callback;
Expand Down
Loading