Skip to content

Commit

Permalink
[Core] Race Conditions: Protect CDataWriter / CDataReader m_event_cal…
Browse files Browse the repository at this point in the history 
…lback_map

Co-authored-by: Rex Schilasky <[email protected]>
  • Loading branch information
@KerstinKeller @rex-schilasky
KerstinKeller and rex-schilasky committed Jan 18, 2024
1 parent eaf1d51 commit d8632e8
Show file tree
Hide file tree
Showing 2 changed files with 61 additions and 42 deletions.
54 changes: 32 additions & 22 deletions ecal/core/src/readwrite/ecal_reader.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -726,6 +726,7 @@ namespace eCAL

// fire sub_event_connected
{
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(sub_event_connected);
if (iter != m_event_callback_map.end())
{
Expand All @@ -741,15 +742,18 @@ namespace eCAL
}

// fire sub_event_update_connection
auto iter = m_event_callback_map.find(sub_event_update_connection);
if (iter != m_event_callback_map.end())
{
data.type = sub_event_update_connection;
data.tid = tid_;
data.ttype = Util::CombinedTopicEncodingAndType(data_type_info_.encoding, data_type_info_.name);
data.tdesc = data_type_info_.descriptor;
data.tdatatype = data_type_info_;
(iter->second)(m_topic_name.c_str(), &data);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(sub_event_update_connection);
if (iter != m_event_callback_map.end())
{
data.type = sub_event_update_connection;
data.tid = tid_;
data.ttype = Util::CombinedTopicEncodingAndType(data_type_info_.encoding, data_type_info_.name);
data.tdesc = data_type_info_.descriptor;
data.tdatatype = data_type_info_;
(iter->second)(m_topic_name.c_str(), &data);
}
}
}

Expand All @@ -760,14 +764,17 @@ namespace eCAL
m_connected = false;

// fire sub_event_disconnected
auto iter = m_event_callback_map.find(sub_event_disconnected);
if (iter != m_event_callback_map.end())
{
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);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(sub_event_disconnected);
if (iter != m_event_callback_map.end())
{
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);
}
}
}
}
Expand Down Expand Up @@ -827,14 +834,17 @@ namespace eCAL
Logging::Log(log_level_warning, msg);
#endif
// we fire the message drop event
auto citer = m_event_callback_map.find(sub_event_dropped);
if (citer != m_event_callback_map.end())
{
SSubEventCallbackData data;
data.type = sub_event_dropped;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = current_clock_;
(citer->second)(m_topic_name.c_str(), &data);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto citer = m_event_callback_map.find(sub_event_dropped);
if (citer != m_event_callback_map.end())
{
SSubEventCallbackData data;
data.type = sub_event_dropped;
data.time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
data.clock = current_clock_;
(citer->second)(m_topic_name.c_str(), &data);
}
}
// increase the drop counter
m_message_drops += clock_difference;
Expand Down
49 changes: 29 additions & 20 deletions ecal/core/src/readwrite/ecal_writer.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1004,25 +1004,31 @@ namespace eCAL
m_connected = true;

// fire pub_event_connected
auto iter = m_event_callback_map.find(pub_event_connected);
if (iter != m_event_callback_map.end())
{
data.type = pub_event_connected;
(iter->second)(m_topic_name.c_str(), &data);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(pub_event_connected);
if (iter != m_event_callback_map.end())
{
data.type = pub_event_connected;
(iter->second)(m_topic_name.c_str(), &data);
}
}
}

// fire pub_event_update_connection
auto iter = m_event_callback_map.find(pub_event_update_connection);
if (iter != m_event_callback_map.end())
{
data.type = pub_event_update_connection;
data.tid = tid_;
// Remove with eCAL6 (next two lines)
data.ttype = Util::CombinedTopicEncodingAndType(tinfo_.encoding, tinfo_.name);
data.tdesc = tinfo_.descriptor;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(pub_event_update_connection);
if (iter != m_event_callback_map.end())
{
data.type = pub_event_update_connection;
data.tid = tid_;
// Remove with eCAL6 (next two lines)
data.ttype = Util::CombinedTopicEncodingAndType(tinfo_.encoding, tinfo_.name);
data.tdesc = tinfo_.descriptor;
data.tdatatype = tinfo_;
(iter->second)(m_topic_name.c_str(), &data);
}
}
}

Expand All @@ -1033,14 +1039,17 @@ namespace eCAL
m_connected = false;

// fire pub_event_disconnected
auto iter = m_event_callback_map.find(pub_event_disconnected);
if (iter != m_event_callback_map.end())
{
SPubEventCallbackData data;
data.type = pub_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);
const std::lock_guard<std::mutex> lock(m_event_callback_map_sync);
auto iter = m_event_callback_map.find(pub_event_disconnected);
if (iter != m_event_callback_map.end())
{
SPubEventCallbackData data;
data.type = pub_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);
}
}
}
}
Expand Down

1 comment on commit d8632e8

@FlorianReimold
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Part of PR #1312

Please sign in to comment.