Fix GAP messages are not sent when there is no Reader requesting the DATA (#6181)

* Refs #23919. Tests

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

* Refs #23919. Fix

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

* Refs #23919. Fix cornercase

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

* Refs #23919. Apply suggestions

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

* Refs #23919. Fix jobs needs log

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

* Refs #23919. Apply suggestions

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

---------

Signed-off-by: Ricardo González Moreno <ricardo@richiware.dev>
(cherry picked from commit 062258c)

# Conflicts:
#	examples/cpp/content_filter/SubscriberApp.hpp
#	src/cpp/rtps/writer/StatefulWriter.cpp
#	test/blackbox/common/DDSBlackboxTestsContentFilter.cpp
#	test/dds/xtypes/TypeLookupServiceSubscriber.cpp

Author: richiware

.github/workflows/reusable-sanitizers-ci.yml

@@ -221,6 +221,7 @@ jobs:
         continue-on-error: true
         uses: eProsima/eProsima-CI/multiplatform/colcon_test@v0
         with:
+          colcon_args_default: ''
           colcon_test_args: ${{ inputs.colcon_test_args }}
           colcon_test_args_default: '--event-handlers=console_direct+ --return-code-on-test-failure'
           ctest_args: ${{ inputs.ctest_args }}
@@ -319,6 +320,7 @@ jobs:
         continue-on-error: true
         uses: eProsima/eProsima-CI/multiplatform/colcon_test@v0
         with:
+          colcon_args_default: ''
           colcon_test_args: ${{ inputs.colcon_test_args }}
           colcon_test_args_default: '--event-handlers=console_direct+ --return-code-on-test-failure'
           ctest_args: ${{ inputs.ctest_args }}
@@ -464,6 +466,7 @@ jobs:
         continue-on-error: true
         uses: eProsima/eProsima-CI/multiplatform/colcon_test@v0
         with:
+          colcon_args_default: ''
           colcon_test_args: ${{ inputs.colcon_test_args }}
           colcon_test_args_default: '--event-handlers=console_direct+ --return-code-on-test-failure'
           ctest_args: ${{ inputs.ctest_args }}

examples/cpp/content_filter/SubscriberApp.hpp

@@ -0,0 +1,119 @@
+// Copyright 2024 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * @file SubscriberApp.hpp
+ *
+ */
+
+#ifndef FASTDDS_EXAMPLES_CPP_CONTENT_FILTER__SUBSCRIBERAPP_HPP
+#define FASTDDS_EXAMPLES_CPP_CONTENT_FILTER__SUBSCRIBERAPP_HPP
+
+#include <atomic>
+#include <condition_variable>
+
+#include <fastdds/dds/core/status/SubscriptionMatchedStatus.hpp>
+#include <fastdds/dds/domain/DomainParticipant.hpp>
+#include <fastdds/dds/subscriber/DataReader.hpp>
+#include <fastdds/dds/subscriber/DataReaderListener.hpp>
+#include <fastdds/dds/subscriber/SampleInfo.hpp>
+#include <fastdds/dds/subscriber/Subscriber.hpp>
+#include <fastdds/dds/topic/ContentFilteredTopic.hpp>
+#include <fastdds/dds/topic/Topic.hpp>
+#include <fastdds/dds/topic/TypeSupport.hpp>
+
+#include "Application.hpp"
+#include "CLIParser.hpp"
+#include "CustomContentFilterFactory.hpp"
+#include "HelloWorld.hpp"
+
+using namespace eprosima::fastdds::dds;
+namespace eprosima {
+namespace fastdds {
+namespace examples {
+namespace content_filter {
+class SubscriberApp : public Application, public DataReaderListener
+{
+public:
+
+    //! Constructor
+    SubscriberApp(
+            const CLIParser::subscriber_config& config,
+            const std::string& topic_name);
+
+    //! Destructor
+    ~SubscriberApp();
+
+    //! Subscription callback
+    void on_data_available(
+            DataReader* reader) override;
+
+    //! Subscriber matched method
+    void on_subscription_matched(
+            DataReader* reader,
+            const SubscriptionMatchedStatus& info) override;
+
+    void on_sample_lost(
+            DataReader*,
+            const SampleLostStatus& status) override
+    {
+        std::cout << "Sample lost: " << status.total_count << std::endl;
+    }
+
+    //! Run subscriber
+    void run() override;
+
+    //! Trigger the end of execution
+    void stop() override;
+
+private:
+
+    //! Return the current state of execution
+    bool is_stopped();
+
+    HelloWorld hello_;
+
+    DomainParticipant* participant_;
+
+    Subscriber* subscriber_;
+
+    Topic* topic_;
+
+    DataReader* reader_;
+
+    TypeSupport type_;
+
+    uint16_t received_samples_;
+
+    uint16_t samples_;
+
+    //! DDS ContentFilteredTopic pointer
+    ContentFilteredTopic* filter_topic_;
+
+    //! Custom filter factory
+    CustomContentFilterFactory filter_factory;
+
+    std::atomic<bool> stop_;
+
+    mutable std::mutex terminate_cv_mtx_;
+
+    std::condition_variable terminate_cv_;
+};
+
+} // namespace content_filter
+} // namespace examples
+} // namespace fastdds
+} // namespace eprosima
+
+#endif // FASTDDS_EXAMPLES_CPP_CONTENT_FILTER__SUBSCRIBERAPP_HPP

include/fastdds/rtps/writer/ReaderProxy.h

@@ -143,7 +143,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.
@@ -345,11 +345,38 @@ class ReaderProxy
      * Get the highest fully acknowledged sequence number.
      * @return the highest fully acknowledged sequence number.
      */
-    SequenceNumber_t changes_low_mark() const
+    inline SequenceNumber_t changes_low_mark() const
     {
         return changes_low_mark_;
     }
 
+    /*!
+     * Get the first sequence number not relevant that was removed without reader being informed.
+     * @return First sequence number.
+     */
+    inline SequenceNumber_t first_irrelevant_removed() const
+    {
+        return first_irrelevant_removed_;
+    }
+
+    /*!
+     * Get the last sequence number not relevant that was removed without reader being informed.
+     * @return last sequence number.
+     */
+    inline SequenceNumber_t last_irrelevant_removed() const
+    {
+        return last_irrelevant_removed_;
+    }
+
+    /*!
+     * Reset the interval of sequence numbers not relevant that were removed without reader being informed.
+     */
+    inline 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.
@@ -450,8 +477,14 @@ class ReaderProxy
     //! Last  NACKFRAG count.
     uint32_t last_nackfrag_count_;
 
+    //! Sequence number of the lowest change not fully acknowledged.
     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;

include/fastdds/rtps/writer/StatefulWriter.h

@@ -486,7 +486,13 @@ class StatefulWriter : public RTPSWriter
      */
     bool ack_timer_expired();
 
-    void send_heartbeat_to_all_readers();
+    /*!
+     * Send heartbeat to all the remote readers.
+     * @param force_separating True to send the heartbeat separately for each reader.
+     * False to send a unique heartbeat to all the readers.
+     */
+    void send_heartbeat_to_all_readers(
+            bool force_separating);
 
     void deliver_sample_to_intraprocesses(
             CacheChange_t* change);
@@ -503,6 +509,10 @@ class StatefulWriter : public RTPSWriter
     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.
      *

src/cpp/rtps/writer/ReaderProxy.cpp

@@ -227,25 +227,40 @@ void ReaderProxy::add_change(
         const ChangeForReader_t& change,
         bool is_relevant)
 {
-    assert(change.getSequenceNumber() > changes_low_mark_);
+    SequenceNumber_t seq_num {change.getSequenceNumber()};
+    assert(seq_num > changes_low_mark_);
     assert(changes_for_reader_.empty() ? true :
-            change.getSequenceNumber() > changes_for_reader_.back().getSequenceNumber());
+            seq_num > changes_for_reader_.back().getSequenceNumber());
 
     // 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_ = seq_num;
+                    last_irrelevant_removed_ = seq_num;
+                }
+                else if  (seq_num == last_irrelevant_removed_ + 1)
+                {
+                    last_irrelevant_removed_ = seq_num;
+                }
+            }
+        }
+        else if (changes_low_mark_ + 1 == seq_num)
         {
-            changes_low_mark_ = change.getSequenceNumber();
+            changes_low_mark_ = seq_num;
         }
         return;
     }
 
     if (changes_for_reader_.push_back(change) == nullptr)
     {
         // This should never happen
-        EPROSIMA_LOG_ERROR(RTPS_READER_PROXY, "Error adding change " << change.getSequenceNumber()
+        EPROSIMA_LOG_ERROR(RTPS_READER_PROXY, "Error adding change " << seq_num
                                                                      << " to reader proxy " << guid());
         eprosima::fastdds::dds::Log::Flush();
         assert(false);
@@ -281,7 +296,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 +343,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 +466,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;
+                            }
+                        }
                     }
                 });
     }