Solve Discovery Server race conditions (#5806)

* Refs #23088: Test reconnection when removing participant



* Refs #23088: Solve EDP-PDP queues race condition



* Refs #23088: Solve data UP + data P race condition



* Refs #23088: Abort writer/reader processing if associated participant not alive



* Refs #23088: Apply suggestions



* Refs #23088: Release change when writer/reader insertion in DB failed



* Refs #23088: Match servers after change update



---------

Signed-off-by: cferreiragonz <[email protected]>
Signed-off-by: Juan Lopez Fernandez <[email protected]>
Co-authored-by: juanlofer-eprosima <[email protected]>
Co-authored-by: cferreiragonz <[email protected]>

Author: 3 people

src/cpp/rtps/builtin/discovery/database/DiscoveryDataBase.cpp

@@ -494,9 +494,6 @@ void DiscoveryDataBase::process_pdp_data_queue()
     // Lock(exclusive mode) mutex locally
     std::lock_guard<std::recursive_mutex> guard(mutex_);
 
-    // Swap DATA queues
-    pdp_data_queue_.Swap();
-
     // Process all messages in the queque
     while (!pdp_data_queue_.Empty())
     {
@@ -534,9 +531,6 @@ bool DiscoveryDataBase::process_edp_data_queue()
     // Lock(exclusive mode) mutex locally
     std::lock_guard<std::recursive_mutex> guard(mutex_);
 
-    // Swap DATA queues
-    edp_data_queue_.Swap();
-
     eprosima::fastdds::rtps::CacheChange_t* change;
     std::string topic_name;
 
@@ -660,7 +654,7 @@ void DiscoveryDataBase::match_new_server_(
         }
     }
     // The resources needed for TCP new connections are created during the matching process when the
-    // DATA(p) is receieved by each server.
+    // DATA(p) is received by each server.
 
     // Create virtual endpoints
     create_virtual_endpoints_(participant_prefix);
@@ -788,32 +782,71 @@ void DiscoveryDataBase::update_participant_from_change_(
 {
     fastdds::rtps::GUID_t change_guid = guid_from_change(ch);
 
+    assert(ch->kind == eprosima::fastdds::rtps::ALIVE);
+
+    // If the change corresponds to a previously removed participant (which hasn't yet been removed from the map since
+    // the DATA(Up) is still unacked), update map with new data and behave as if it was a new participant.
+    // Remove also the old change from the disposals collection, if it was added just before
+    if (participant_info.change()->kind != eprosima::fastdds::rtps::ALIVE)
+    {
+        // Update the change data
+        participant_info.participant_change_data(change_data);
+
+        // Remove old change from disposals if it was added just before to avoid sending data UP
+        auto it = std::find(disposals_.begin(), disposals_.end(), participant_info.change());
+        if (it != disposals_.end())
+        {
+            disposals_.erase(it);
+        }
+
+        // Update change. This should add the UNALIVE change to changes_to_release_, which should later both remove the
+        // change from the writer's history and release the change
+        update_change_and_unmatch_(ch, participant_info);
+
+        // If it is local and server we have to create virtual endpoints, except for our own server
+        if (change_guid.guidPrefix != server_guid_prefix_ && !change_data.is_client() && change_data.is_local())
+        {
+            // Match new server and create virtual endpoints
+            // NOTE: match after having updated the change, so virtual endpoints are not discarded for having
+            // an associated unalive participant
+            match_new_server_(change_guid.guidPrefix, change_data.is_superclient());
+        }
+
+        // Treat as a new participant found
+        new_updates_++;
+        if (change_guid.guidPrefix != server_guid_prefix_)
+        {
+            server_acked_by_all(false);
+        }
+    }
+
     // Specific case when a Data(P) from an entity A known as remote comes from the very entity A (we have
     // the Data(P) because of other server B, but now it arrives from A itself)
     // The entity A changes to local
     // Must be local data, or else it is a remote endpoint and should not be changed
-    if (change_guid.guidPrefix != server_guid_prefix_ && change_data.is_local() &&
+    else if (change_guid.guidPrefix != server_guid_prefix_ && change_data.is_local() &&
             DiscoveryDataBase::participant_data_has_changed_(participant_info, change_data))
     {
+        // Update the change data
+        participant_info.participant_change_data(change_data);
+
+        // Update change
+        update_change_and_unmatch_(ch, participant_info);
+
         // If the participant changes to server local, virtual endpoints must be added
         // If it is local and server the only possibility is it was a remote server and it must be converted to local
         if (!change_data.is_client())
         {
+            // NOTE: match after having updated the change in order to send the new Data(P)
             match_new_server_(change_guid.guidPrefix, change_data.is_superclient());
         }
 
-        // Update the change data
-        participant_info.participant_change_data(change_data);
-
-        // Update change
-        update_change_and_unmatch_(ch, participant_info);
-
         // Treat as a new participant found
         new_updates_++;
         server_acked_by_all(false);
 
         // It is possible that this Data(P) is in our history if it has not been acked by all
-        // In this case we have to resent it with the new update
+        // In this case we have to resend it with the new update
         if (!participant_info.is_acked_by_all())
         {
             add_pdp_to_send_(ch);
@@ -916,6 +949,29 @@ void DiscoveryDataBase::create_writers_from_change_(
     // The writer was NOT known by the database
     else
     {
+        // Check if corresponding participant is known, abort otherwise
+        // NOTE: Processing a DATA(w) should always be preceded by the reception and processing of its corresponding
+        // participant. However, one may receive a DATA(w) just after the participant has been removed, case in which the
+        // former should no longer be processed.
+        std::map<eprosima::fastdds::rtps::GuidPrefix_t, DiscoveryParticipantInfo>::iterator writer_part_it =
+                participants_.find(writer_guid.guidPrefix);
+        if (writer_part_it == participants_.end())
+        {
+            EPROSIMA_LOG_ERROR(DISCOVERY_DATABASE,
+                    "Writer " << writer_guid << " has no associated participant. Skipping");
+            assert(topic_name != virtual_topic_);
+            changes_to_release_.push_back(ch); // Release change so it can be reused
+            return;
+        }
+        else if (writer_part_it->second.change()->kind != fastdds::rtps::ChangeKind_t::ALIVE)
+        {
+            EPROSIMA_LOG_WARNING(DISCOVERY_DATABASE,
+                    "Writer " << writer_guid << " is associated to a removed participant. Skipping");
+            assert(topic_name != virtual_topic_);
+            changes_to_release_.push_back(ch); // Release change so it can be reused
+            return;
+        }
+
         // Add entry to writers_
         DiscoveryEndpointInfo tmp_writer(
             ch,
@@ -936,18 +992,7 @@ void DiscoveryDataBase::create_writers_from_change_(
         new_updates_++;
 
         // Add entry to participants_[guid_prefix]::writers
-        std::map<eprosima::fastdds::rtps::GuidPrefix_t, DiscoveryParticipantInfo>::iterator writer_part_it =
-                participants_.find(writer_guid.guidPrefix);
-        if (writer_part_it != participants_.end())
-        {
-            writer_part_it->second.add_writer(writer_guid);
-        }
-        else
-        {
-            EPROSIMA_LOG_ERROR(DISCOVERY_DATABASE,
-                    "Writer " << writer_guid << " has no associated participant. Skipping");
-            return;
-        }
+        writer_part_it->second.add_writer(writer_guid);
 
         // Add writer to writers_by_topic_[topic_name]
         add_writer_to_topic_(writer_guid, topic_name);
@@ -1034,6 +1079,29 @@ void DiscoveryDataBase::create_readers_from_change_(
     // The reader was NOT known by the database
     else
     {
+        // Check if corresponding participant is known, abort otherwise
+        // NOTE: Processing a DATA(r) should always be preceded by the reception and processing of its corresponding
+        // participant. However, one may receive a DATA(r) just after the participant has been removed, case in which the
+        // former should no longer be processed.
+        std::map<eprosima::fastdds::rtps::GuidPrefix_t, DiscoveryParticipantInfo>::iterator reader_part_it =
+                participants_.find(reader_guid.guidPrefix);
+        if (reader_part_it == participants_.end())
+        {
+            EPROSIMA_LOG_ERROR(DISCOVERY_DATABASE,
+                    "Reader " << reader_guid << " has no associated participant. Skipping");
+            assert(topic_name != virtual_topic_);
+            changes_to_release_.push_back(ch); // Release change so it can be reused
+            return;
+        }
+        else if (reader_part_it->second.change()->kind != fastdds::rtps::ChangeKind_t::ALIVE)
+        {
+            EPROSIMA_LOG_WARNING(DISCOVERY_DATABASE,
+                    "Reader " << reader_guid << " is associated to a removed participant. Skipping");
+            assert(topic_name != virtual_topic_);
+            changes_to_release_.push_back(ch); // Release change so it can be reused
+            return;
+        }
+
         // Add entry to readers_
         DiscoveryEndpointInfo tmp_reader(
             ch,
@@ -1054,18 +1122,7 @@ void DiscoveryDataBase::create_readers_from_change_(
         new_updates_++;
 
         // Add entry to participants_[guid_prefix]::readers
-        std::map<eprosima::fastdds::rtps::GuidPrefix_t, DiscoveryParticipantInfo>::iterator reader_part_it =
-                participants_.find(reader_guid.guidPrefix);
-        if (reader_part_it != participants_.end())
-        {
-            reader_part_it->second.add_reader(reader_guid);
-        }
-        else
-        {
-            EPROSIMA_LOG_ERROR(DISCOVERY_DATABASE,
-                    "Reader " << reader_guid << " has no associated participant. Skipping");
-            return;
-        }
+        reader_part_it->second.add_reader(reader_guid);
 
         // Add reader to readers_by_topic_[topic_name]
         add_reader_to_topic_(reader_guid, topic_name);
@@ -1347,7 +1404,7 @@ void DiscoveryDataBase::process_dispose_participant_(
         delete_reader_entity_(reader_guid);
     }
 
-    // All participant endoints must be already unmatched in others endopoints relevant_ack maps
+    // All participant endpoints must be already unmatched in others endpoints relevant_ack maps
 
     // Unmatch own participant
     unmatch_participant_(participant_guid.guidPrefix);
@@ -1618,6 +1675,14 @@ bool DiscoveryDataBase::data_queue_empty()
     return (pdp_data_queue_.BothEmpty() && edp_data_queue_.BothEmpty());
 }
 
+void DiscoveryDataBase::swap_data_queues()
+{
+    // Swap EDP before PDP to avoid race condition in which both data P and w/r are received at the same time,
+    // just after having swapped the PDP queue
+    edp_data_queue_.Swap();
+    pdp_data_queue_.Swap();
+}
+
 bool DiscoveryDataBase::is_participant(
         const eprosima::fastdds::rtps::GUID_t& guid)
 {

src/cpp/rtps/builtin/discovery/database/DiscoveryDataBase.hpp

@@ -294,6 +294,9 @@ class DiscoveryDataBase
     // Check if the data queue is empty
     bool data_queue_empty();
 
+    // Swap both EDP and PDP data queues
+    void swap_data_queues();
+
     void to_json(
             nlohmann::json& j) const;
 

src/cpp/rtps/builtin/discovery/participant/PDPServer.cpp

@@ -1080,6 +1080,9 @@ bool PDPServer::remove_remote_participant(
 bool PDPServer::process_data_queues()
 {
     EPROSIMA_LOG_INFO(RTPS_PDP_SERVER, "process_data_queues start");
+    // Swap both as a first step in order to avoid the following race condition: reception of data w/r while processing
+    // the PDP queue, not having processed yet the corresponding data P (also received while processing the queue).
+    discovery_db_.swap_data_queues();
     discovery_db_.process_pdp_data_queue();
     return discovery_db_.process_edp_data_queue();
 }