Refs #23919. Fix

Signed-off-by: Ricardo González Moreno <ricardo@richiware.dev>

Author: richiware

src/cpp/rtps/writer/ReaderProxy.cpp

@@ -234,8 +234,22 @@ void ReaderProxy::add_change(
     // Irrelevant changes are not added to the collection
     if (!is_relevant)
     {
-        if ( !is_reliable_ &&
-                changes_low_mark_ + 1 == change.getSequenceNumber())
+        if (is_reliable_)
+        {
+            if (!is_local_reader())
+            {
+                if (SequenceNumber_t::unknown() == first_irrelevant_removed_)
+                {
+                    first_irrelevant_removed_ = change.getSequenceNumber();
+                    last_irrelevant_removed_ = change.getSequenceNumber();
+                }
+                else if  (change.getSequenceNumber() == last_irrelevant_removed_ + 1)
+                {
+                    last_irrelevant_removed_ = change.getSequenceNumber();
+                }
+            }
+        }
+        else if (changes_low_mark_ + 1 == change.getSequenceNumber())
         {
             changes_low_mark_ = change.getSequenceNumber();
         }
@@ -281,7 +295,7 @@ bool ReaderProxy::change_is_unsent(
         FragmentNumber_t& next_unsent_frag,
         SequenceNumber_t& gap_seq,
         const SequenceNumber_t& min_seq,
-        bool& need_reactivate_periodic_heartbeat) const
+        bool& need_reactivate_periodic_heartbeat)
 {
     if (seq_num <= changes_low_mark_ || changes_for_reader_.empty())
     {
@@ -328,6 +342,21 @@ bool ReaderProxy::change_is_unsent(
                         gap_seq = SequenceNumber_t::unknown();
                     }
                 }
+
+                if (SequenceNumber_t::unknown() != first_irrelevant_removed_ &&
+                        SequenceNumber_t::unknown() != gap_seq)
+                {
+                    // Check if the hole is due to irrelevant changes removed without informing the reader
+                    if (gap_seq == first_irrelevant_removed_)
+                    {
+                        first_irrelevant_removed_ = SequenceNumber_t::unknown();
+                        last_irrelevant_removed_ = SequenceNumber_t::unknown();
+                    }
+                    else if (gap_seq < last_irrelevant_removed_)
+                    {
+                        last_irrelevant_removed_ = gap_seq - 1;
+                    }
+                }
             }
         }
     }
@@ -436,6 +465,20 @@ bool ReaderProxy::requested_changes_set(
                     else if ((sit >= min_seq_in_history) && (sit > changes_low_mark_))
                     {
                         gap_builder.add(sit);
+
+                        if (SequenceNumber_t::unknown() != first_irrelevant_removed_)
+                        {
+                            // Check if the hole is due to irrelevant changes removed without informing the reader
+                            if (sit == first_irrelevant_removed_)
+                            {
+                                first_irrelevant_removed_ = SequenceNumber_t::unknown();
+                                last_irrelevant_removed_ = SequenceNumber_t::unknown();
+                            }
+                            else if (sit < last_irrelevant_removed_)
+                            {
+                                last_irrelevant_removed_ = sit - 1;
+                            }
+                        }
                     }
                 });
     }

src/cpp/rtps/writer/ReaderProxy.hpp

@@ -139,7 +139,7 @@ class ReaderProxy
             FragmentNumber_t& next_unsent_frag,
             SequenceNumber_t& gap_seq,
             const SequenceNumber_t& min_seq,
-            bool& need_reactivate_periodic_heartbeat) const;
+            bool& need_reactivate_periodic_heartbeat);
 
     /**
      * Mark all changes up to the one indicated by seq_num as Acknowledged.
@@ -346,6 +346,22 @@ class ReaderProxy
         return changes_low_mark_;
     }
 
+    SequenceNumber_t first_irrelevant_removed() const
+    {
+        return first_irrelevant_removed_;
+    }
+
+    SequenceNumber_t last_irrelevant_removed() const
+    {
+        return last_irrelevant_removed_;
+    }
+
+    void reset_irrelevant_removed()
+    {
+        first_irrelevant_removed_ = SequenceNumber_t::unknown();
+        last_irrelevant_removed_ = SequenceNumber_t::unknown();
+    }
+
     /**
      * Change the interval of nack-supression event.
      * @param interval Time from data sending to acknack processing.
@@ -448,6 +464,11 @@ class ReaderProxy
 
     SequenceNumber_t changes_low_mark_;
 
+    //! First sequence number not relevant that was removed without reader being informed.
+    SequenceNumber_t first_irrelevant_removed_ {SequenceNumber_t::unknown()};
+    //! Last sequence number not relevant that was removed without reader being informed.
+    SequenceNumber_t last_irrelevant_removed_ {SequenceNumber_t::unknown()};
+
     bool active_ = false;
 
     using ChangeIterator = ResourceLimitedVector<ChangeForReader_t, std::true_type>::iterator;

src/cpp/rtps/writer/StatefulWriter.cpp

@@ -520,11 +520,12 @@ bool StatefulWriter::change_removed_by_history(
     return ret_value;
 }
 
-void StatefulWriter::send_heartbeat_to_all_readers()
+void StatefulWriter::send_heartbeat_to_all_readers(
+        bool force_separating)
 {
     // This method is only called from send_periodic_heartbeat
 
-    if (separate_sending_enabled_)
+    if (separate_sending_enabled_ || force_separating)
     {
         for (ReaderProxy* reader : matched_remote_readers_)
         {
@@ -664,6 +665,14 @@ DeliveryRetCode StatefulWriter::deliver_sample_to_network(
                 should_be_sent = true;
                 inline_qos |= (*remote_reader)->expects_inline_qos();
 
+                if (SequenceNumber_t::unknown() != (*remote_reader)->first_irrelevant_removed())
+                {
+                    // Send GAP with irrelevant changes that are not in history.
+                    group.sender(this, (*remote_reader)->message_sender());
+                    add_gaps_for_removed_irrelevants(**remote_reader, group);
+                    group.sender(this, &locator_selector);         // This makes the flush_and_reset().
+                }
+
                 // If there is a hole (removed from history or not relevants) between previous sample and this one,
                 // send it a personal GAP.
                 if (SequenceNumber_t::unknown() != gap_seq)
@@ -1686,7 +1695,8 @@ bool StatefulWriter::send_periodic_heartbeat(
     std::lock_guard<RecursiveTimedMutex> guardW(mp_mutex);
     std::lock_guard<LocatorSelectorSender> guard_locator_selector_general(locator_selector_general_);
 
-    bool unacked_changes = false;
+    bool unacked_changes {false};
+    bool irrelevants_removed {false};
     if (!liveliness)
     {
         SequenceNumber_t first_seq_to_check_acknowledge = get_seq_num_min();
@@ -1695,20 +1705,30 @@ bool StatefulWriter::send_periodic_heartbeat(
             first_seq_to_check_acknowledge = history_->next_sequence_number() - 1;
         }
 
-        unacked_changes = for_matched_readers(matched_local_readers_, matched_datasharing_readers_,
-                        matched_remote_readers_,
-                        [first_seq_to_check_acknowledge](ReaderProxy* reader)
-                        {
-                            return reader->has_unacknowledged(first_seq_to_check_acknowledge);
-                        }
-                        );
+        for_matched_readers(matched_local_readers_, matched_datasharing_readers_,
+                matched_remote_readers_,
+                [first_seq_to_check_acknowledge, &unacked_changes, &irrelevants_removed](ReaderProxy* reader)
+                {
+                    if (!unacked_changes)
+                    {
+                        unacked_changes = reader->has_unacknowledged(first_seq_to_check_acknowledge);
+                    }
+
+                    if (!irrelevants_removed)
+                    {
+                        irrelevants_removed = SequenceNumber_t::unknown() != reader->first_irrelevant_removed();
+                    }
+
+                    return unacked_changes && irrelevants_removed;
+                }
+                );
 
         if (unacked_changes)
         {
             try
             {
                 //TODO if separating, here sends periodic for all readers, instead of ones needed it.
-                send_heartbeat_to_all_readers();
+                send_heartbeat_to_all_readers(irrelevants_removed);
             }
             catch (const RTPSMessageGroup::timeout&)
             {
@@ -1798,6 +1818,7 @@ void StatefulWriter::send_heartbeat_to_nts(
                     assert(firstSeq <= lastSeq);
                     if (!liveliness)
                     {
+                        add_gaps_for_removed_irrelevants(remoteReaderProxy, group);
                         add_gaps_for_holes_in_history(group);
                     }
                 }
@@ -2247,6 +2268,19 @@ bool StatefulWriter::get_connections(
 
 #endif // ifdef FASTDDS_STATISTICS
 
+void StatefulWriter::add_gaps_for_removed_irrelevants(
+        ReaderProxy& remoteReaderProxy,
+        RTPSMessageGroup& group)
+{
+    if (SequenceNumber_t::unknown() != remoteReaderProxy.first_irrelevant_removed())
+    {
+        group.add_gap(remoteReaderProxy.first_irrelevant_removed(),
+                SequenceNumberSet_t(remoteReaderProxy.last_irrelevant_removed() + 1),
+                remoteReaderProxy.guid());
+        remoteReaderProxy.reset_irrelevant_removed();
+    }
+}
+
 void StatefulWriter::add_gaps_for_holes_in_history(
         RTPSMessageGroup& group)
 {

src/cpp/rtps/writer/StatefulWriter.hpp

@@ -385,7 +385,8 @@ class StatefulWriter : public BaseWriter
      */
     bool ack_timer_expired();
 
-    void send_heartbeat_to_all_readers();
+    void send_heartbeat_to_all_readers(
+            bool force_separating);
 
     void deliver_sample_to_intraprocesses(
             CacheChange_t* change);
@@ -402,6 +403,10 @@ class StatefulWriter : public BaseWriter
     void prepare_datasharing_delivery(
             CacheChange_t* change);
 
+    void add_gaps_for_removed_irrelevants(
+            ReaderProxy& remoteReaderProxy,
+            RTPSMessageGroup& group);
+
     /**
      * Check the StatefulWriter's sequence numbers and add the required GAP messages to the provided message group.
      *