Skip to content

Commit

Permalink
[core] new-datarw-connection-logic (#1729)
Browse files Browse the repository at this point in the history 
- connection state logic from datareader/writer (sub/pub) corrected to get IsPublished/IsSubscribed (and the connection events) only if the connection is fully established
- connection state need now a second registration loop to get fully active
- Send(), Receive() functions are not affected by the new state handling
- test added to check correct IsPublished() and IsSubscribed() behavior
  • Loading branch information
@rex-schilasky
rex-schilasky authored Sep 6, 2024
1 parent 8ae8323 commit 9ba6e21
Show file tree
Hide file tree
Showing 10 changed files with 541 additions and 319 deletions.
26 changes: 16 additions & 10 deletions cpack/innosetup/ecal_setup.iss.in
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,13 +75,15 @@ Source: "{#ComponentStagingDir}\runtime\*"; DestDir: "{app}"; Fla
Source: "{#ComponentStagingDir}\Unspecified\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: runtime
Source: "{#ComponentStagingDir}\libraries\bin\*"; DestDir: "{app}\bin\"; Flags: ignoreversion recursesubdirs; Components: runtime

; Source: "{#ComponentStagingDir}\_configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: not replaceconf; Flags: ignoreversion confirmoverwrite; Components: runtime
; Source: "{#ComponentStagingDir}\_configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: replaceconf; Flags: ignoreversion; Components: runtime
; Source: "{#ComponentStagingDir}\_configuration\cfg\ecaltime.ini"; DestDir: "{commonappdata}\eCAL\"; Flags: ignoreversion; Components: runtime
; we need to change the location of these configuration file (as hotfix we use skipifsourcedoesntexist flag)
Source: "{#ComponentStagingDir}\configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: not replaceconf; Flags: ignoreversion skipifsourcedoesntexist confirmoverwrite; Components: runtime
Source: "{#ComponentStagingDir}\configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: replaceconf; Flags: ignoreversion skipifsourcedoesntexist; Components: runtime
Source: "{#ComponentStagingDir}\configuration\cfg\ecaltime.ini"; DestDir: "{commonappdata}\eCAL\"; Flags: ignoreversion skipifsourcedoesntexist; Components: runtime

Source: "{#ComponentStagingDir}\configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: not replaceconf; Flags: ignoreversion confirmoverwrite; Components: runtime
Source: "{#ComponentStagingDir}\configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: replaceconf; Flags: ignoreversion; Components: runtime
Source: "{#ComponentStagingDir}\configuration\cfg\ecaltime.ini"; DestDir: "{commonappdata}\eCAL\"; Flags: ignoreversion; Components: runtime
; we need to change the location of these configuration file (as hotfix we use skipifsourcedoesntexist flag)
Source: "{#ComponentStagingDir}\_configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: not replaceconf; Flags: ignoreversion skipifsourcedoesntexist confirmoverwrite; Components: runtime
Source: "{#ComponentStagingDir}\_configuration\cfg\ecal.yaml"; DestDir: "{commonappdata}\eCAL\"; Tasks: replaceconf; Flags: ignoreversion skipifsourcedoesntexist; Components: runtime
Source: "{#ComponentStagingDir}\_configuration\cfg\ecaltime.ini"; DestDir: "{commonappdata}\eCAL\"; Flags: ignoreversion skipifsourcedoesntexist; Components: runtime

; applications
Source: "{#ComponentStagingDir}\app\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: applications
Expand Down Expand Up @@ -122,13 +124,17 @@ Source: "{#DebugSdkStagingDir}\protobuf-export\*"; DestDir: "{app}";
; Source: "{#DebugSdkStagingDir}\protoc\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\protobuf

; sdk\hdf5
; Source: "{#ComponentStagingDir}\_configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
Source: "{#ComponentStagingDir}\configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
; we need to change the location of these configuration file (as hotfix we use skipifsourcedoesntexist flag)
Source: "{#ComponentStagingDir}\configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs skipifsourcedoesntexist; Components: sdk\hdf5
Source: "{#ComponentStagingDir}\_configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs skipifsourcedoesntexist; Components: sdk\hdf5

Source: "{#ComponentStagingDir}\headers\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
Source: "{#ComponentStagingDir}\libraries\lib\*"; DestDir: "{app}\lib"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5

; Source: "{#DebugSdkStagingDir}\_configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
Source: "{#DebugSdkStagingDir}\configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
; we need to change the location of these configuration file (as hotfix we use skipifsourcedoesntexist flag)
Source: "{#DebugSdkStagingDir}\configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs skipifsourcedoesntexist; Components: sdk\hdf5
Source: "{#DebugSdkStagingDir}\_configinstall\*"; DestDir: "{app}"; Flags: ignoreversion recursesubdirs skipifsourcedoesntexist; Components: sdk\hdf5

Source: "{#DebugSdkStagingDir}\libraries\lib\*"; DestDir: "{app}\lib"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5
Source: "{#DebugSdkStagingDir}\libraries\bin\*"; DestDir: "{app}\bin\"; Flags: ignoreversion recursesubdirs; Components: sdk\hdf5

Expand Down
11 changes: 1 addition & 10 deletions ecal/core/src/pubsub/ecal_subgate.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -256,16 +256,7 @@ namespace eCAL
{
iter->second->ApplyLayerParameter(publication_info, tlayer.type, tlayer.par_layer);
}
// we only inform the subscriber when the publisher has already recognized at least one subscriber
// this should avoid to set the "IsPublished" state before the publisher is able to send data
const bool local_publication = publication_info.host_name == Process::GetHostName();
const bool external_publication = !local_publication;
const bool local_confirmed = local_publication && (ecal_sample_.topic.connections_loc > 0);
const bool external_confirmed = external_publication && (ecal_sample_.topic.connections_ext > 0);
if(local_confirmed || external_confirmed)
{
iter->second->ApplyPublication(publication_info, topic_information, layer_states);
}
iter->second->ApplyPublication(publication_info, topic_information, layer_states);
}
}

Expand Down
195 changes: 127 additions & 68 deletions ecal/core/src/readwrite/ecal_reader.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,6 @@ namespace eCAL
m_topic_info(topic_info_),
m_topic_size(0),
m_config(config_),
m_connected(false),
m_receive_time(0),
m_clock(0),
m_frequency_calculator(3.0f),
Expand Down Expand Up @@ -277,35 +276,98 @@ namespace eCAL
m_id_set = id_set_;
}

void CDataReader::ApplyPublication(const SPublicationInfo& publication_info_, const SDataTypeInformation& data_type_info_, const SLayerStates& layer_states_)
void CDataReader::ApplyPublication(const SPublicationInfo& publication_info_, const SDataTypeInformation& data_type_info_, const SLayerStates& pub_layer_states_)
{
// flag write enabled from publisher side (information not used yet)
#if ECAL_CORE_TRANSPORT_UDP
m_layers.udp.write_enabled = layer_states_.udp.write_enabled;
m_layers.udp.write_enabled = pub_layer_states_.udp.write_enabled;
#endif
#if ECAL_CORE_TRANSPORT_SHM
m_layers.shm.write_enabled = layer_states_.shm.write_enabled;
m_layers.shm.write_enabled = pub_layer_states_.shm.write_enabled;
#endif
#if ECAL_CORE_TRANSPORT_TCP
m_layers.tcp.write_enabled = layer_states_.tcp.write_enabled;
m_layers.tcp.write_enabled = pub_layer_states_.tcp.write_enabled;
#endif

FireConnectEvent(publication_info_.entity_id, data_type_info_);
// add key to connection map, including connection state
bool is_new_connection = false;
bool is_updated_connection = false;
{
const std::lock_guard<std::mutex> lock(m_connection_map_mtx);
auto publication_info_iter = m_connection_map.find(publication_info_);

if (publication_info_iter == m_connection_map.end())
{
// add publisher to connection map, connection state false
m_connection_map[publication_info_] = SConnection{ data_type_info_, pub_layer_states_, false };
}
else
{
// existing connection, we got the second update now
auto& connection = publication_info_iter->second;

// if this connection was inactive before
// activate it now and flag a new connection finally
if (!connection.state)
{
is_new_connection = true;
}
// the connection was active, so we just update it
else
{
is_updated_connection = true;
}

// update the data type and layer states, even if the connection is not new
connection = SConnection{ data_type_info_, pub_layer_states_, true };
}

// update connection count
m_connection_count = GetConnectionCount();
}

// add key to publisher map
// handle these events outside the lock
if (is_new_connection)
{
const std::lock_guard<std::mutex> lock(m_pub_map_mtx);
m_pub_map[publication_info_] = std::make_tuple(data_type_info_, layer_states_);
// fire connect event
FireConnectEvent(publication_info_.entity_id, data_type_info_);
}
else if (is_updated_connection)
{
// fire update event
FireUpdateEvent(publication_info_.entity_id, data_type_info_);
}

#ifndef NDEBUG
// log it
Logging::Log(log_level_debug3, m_topic_name + "::CDataReader::ApplyPublication");
#endif
}

void CDataReader::RemovePublication(const SPublicationInfo& publication_info_)
{
// remove key from publisher map
// remove key from connection map
bool last_connection_gone(false);
{
const std::lock_guard<std::mutex> lock(m_pub_map_mtx);
m_pub_map.erase(publication_info_);
const std::lock_guard<std::mutex> lock(m_connection_map_mtx);

m_connection_map.erase(publication_info_);
last_connection_gone = m_connection_map.empty();

// update connection count
m_connection_count = GetConnectionCount();
}

if (last_connection_gone)
{
// fire disconnect event
FireDisconnectEvent();
}

#ifndef NDEBUG
// log it
Logging::Log(log_level_debug3, m_topic_name + "::CDataReader::RemovePublication");
#endif
}

void CDataReader::ApplyLayerParameter(const SPublicationInfo& publication_info_, eTLayerType type_, const Registration::ConnectionPar& parameter_)
Expand Down Expand Up @@ -540,23 +602,20 @@ namespace eCAL
#endif // ECAL_CORE_REGISTRATION
}

void CDataReader::CheckConnections()
Registration::Sample CDataReader::GetRegistration()
{
const std::lock_guard<std::mutex> lock(m_pub_map_mtx);

if (m_pub_map.empty())
{
FireDisconnectEvent();
}
// return registration
return GetRegistrationSample();
}

Registration::Sample CDataReader::GetRegistration()
bool CDataReader::IsPublished() const
{
// check connection timeouts
CheckConnections();
return m_connection_count > 0;
}

// return registration
return GetRegistrationSample();
size_t CDataReader::GetPublisherCount() const
{
return m_connection_count;
}

Registration::Sample CDataReader::GetRegistrationSample()
Expand Down Expand Up @@ -732,62 +791,62 @@ namespace eCAL

void CDataReader::FireConnectEvent(const std::string& tid_, const SDataTypeInformation& tinfo_)
{
SSubEventCallbackData data;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = 0;

if (!m_connected)
{
m_connected = true;

// fire sub_event_connected
{
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_connected);
if (iter != m_event_callback_map.end() && iter->second)
{
data.type = sub_event_connected;
data.tid = tid_;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
}
}
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_connected);
if (iter != m_event_callback_map.end() && iter->second)
{
SSubEventCallbackData data;
data.type = sub_event_connected;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = 0;
data.tid = tid_;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
}
}

// fire sub_event_update_connection
{
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_update_connection);
if (iter != m_event_callback_map.end() && iter->second)
{
data.type = sub_event_update_connection;
data.tid = tid_;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
}
void CDataReader::FireUpdateEvent(const std::string& tid_, const SDataTypeInformation& tinfo_)
{
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_update_connection);
if (iter != m_event_callback_map.end() && iter->second)
{
SSubEventCallbackData data;
data.type = sub_event_update_connection;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = 0;
data.tid = tid_;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
}
}

void CDataReader::FireDisconnectEvent()
{
if (m_connected)
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_disconnected);
if (iter != m_event_callback_map.end() && iter->second)
{
m_connected = false;
SSubEventCallbackData data;
data.type = sub_event_disconnected;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = 0;
(iter->second)(m_topic_name.c_str(), &data);
}
}

// fire sub_event_disconnected
size_t CDataReader::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)
{
const std::lock_guard<std::mutex> lock(m_event_callback_map_mtx);
auto iter = m_event_callback_map.find(sub_event_disconnected);
if (iter != m_event_callback_map.end() && iter->second)
{
SSubEventCallbackData data;
data.type = sub_event_disconnected;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = 0;
(iter->second)(m_topic_name.c_str(), &data);
}
count++;
}
}
return count;
}

bool CDataReader::CheckMessageClock(const std::string& tid_, long long current_clock_)
Expand Down
35 changes: 16 additions & 19 deletions ecal/core/src/readwrite/ecal_reader.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
#include <queue>
#include <set>
#include <string>
#include <tuple>
#include <unordered_map>

namespace eCAL
Expand Down Expand Up @@ -82,25 +81,16 @@ namespace eCAL

void SetID(const std::set<long long>& id_set_);

void ApplyPublication(const SPublicationInfo& publication_info_, const SDataTypeInformation& data_type_info_, const SLayerStates& layer_states_);
void ApplyPublication(const SPublicationInfo& publication_info_, const SDataTypeInformation& data_type_info_, const SLayerStates& pub_layer_states_);
void RemovePublication(const SPublicationInfo& publication_info_);

void ApplyLayerParameter(const SPublicationInfo& publication_info_, eTLayerType type_, const Registration::ConnectionPar& parameter_);

Registration::Sample GetRegistration();
bool IsCreated() const { return(m_created); }

bool IsPublished() const
{
std::lock_guard<std::mutex> const lock(m_pub_map_mtx);
return(!m_pub_map.empty());
}

size_t GetPublisherCount() const
{
const std::lock_guard<std::mutex> lock(m_pub_map_mtx);
return(m_pub_map.size());
}
bool IsPublished() const;
size_t GetPublisherCount() const;

std::string GetTopicName() const { return(m_topic_name); }
std::string GetTopicID() const { return(m_topic_id); }
Expand All @@ -115,16 +105,17 @@ namespace eCAL
void Register();
void Unregister();

void CheckConnections();

Registration::Sample GetRegistrationSample();
Registration::Sample GetUnregistrationSample();

void StartTransportLayer();
void StopTransportLayer();

void FireConnectEvent(const std::string& tid_, const SDataTypeInformation& tinfo_);
void FireUpdateEvent(const std::string& tid_, const SDataTypeInformation& tinfo_);
void FireDisconnectEvent();

size_t GetConnectionCount();

bool CheckMessageClock(const std::string& tid_, long long current_clock_);

Expand All @@ -141,10 +132,16 @@ namespace eCAL
std::atomic<size_t> m_topic_size;
Subscriber::Configuration m_config;

std::atomic<bool> m_connected;
using PublicationMapT = std::map<SPublicationInfo, std::tuple<SDataTypeInformation, SLayerStates>>;
mutable std::mutex m_pub_map_mtx;
PublicationMapT m_pub_map;
struct SConnection
{
SDataTypeInformation data_type_info;
SLayerStates layer_states;
bool state = false;
};
using PublicationMapT = std::map<SPublicationInfo, SConnection>;
mutable std::mutex m_connection_map_mtx;
PublicationMapT m_connection_map;
std::atomic<size_t> m_connection_count{ 0 };

mutable std::mutex m_read_buf_mtx;
std::condition_variable m_read_buf_cv;
Expand Down
Loading

0 comments on commit 9ba6e21

Please sign in to comment.