[20815] Only apply content filter to ALIVE changes (#4876)

* Refs #20815: Add regression test Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Only apply filter to ALIVE changes Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Rename change_is_relevant_for_filter argument Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Refactor test Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Cast loop index to uint16_t for assigning it to the key field Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Refactor test so PubSubWriter can be used directly Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Fix memory leak Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Apply Mario's suggestions Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Add type traits to PubSubWriterReader and PubSubParticipant Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Default move ctor and assignment in DynamicLoanableSequence Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Add doxygen in DynamicLoanableSequence Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Delete copy semantic from DynamicLoanableSequence Signed-off-by: eduponz <eduardoponz@eprosima.com> * Refs #20815: Use alias for TypeTraits::DataListType in PubSub* classes Signed-off-by: eduponz <eduardoponz@eprosima.com> --------- Signed-off-by: eduponz <eduardoponz@eprosima.com>
eProsima · Jun 6, 2024 · 9a64956 · 9a64956
1 parent 43f39bd
commit 9a64956
Show file tree

Hide file tree

Showing 17 changed files with 1,009 additions and 101 deletions.
diff --git a/include/fastrtps/types/DynamicLoanableSequence.hpp b/include/fastrtps/types/DynamicLoanableSequence.hpp
@@ -0,0 +1,184 @@
+// 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.
+
+#ifndef _TYPES_DYNAMIC_LOANABLE_SEQUENCE_HPP_
+#define _TYPES_DYNAMIC_LOANABLE_SEQUENCE_HPP_
+
+#include <cassert>
+#include <memory>
+
+#include <fastdds/dds/core/LoanableSequence.hpp>
+#include <fastdds/dds/log/Log.hpp>
+#include <fastrtps/types/DynamicData.h>
+#include <fastrtps/types/DynamicPubSubType.h>
+#include <fastrtps/types/DynamicTypePtr.h>
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+
+/**
+ * @brief LoanableSequence specialization for DynamicData.
+ *
+ * This class provides a sequence container for handling loanable collections
+ * of DynamicData.
+ *
+ * @tparam _NonConstEnabler to enable data access through [] operator on non-const sequences.
+ */
+template<typename _NonConstEnabler>
+class LoanableSequence<fastrtps::types::DynamicData, _NonConstEnabler>
+    : public LoanableTypedCollection<fastrtps::types::DynamicData, _NonConstEnabler>
+{
+public:
+
+    /// Type for the size of the sequence.
+    using size_type = LoanableCollection::size_type;
+
+    /// Type for the elements in the sequence.
+    using element_type = LoanableCollection::element_type;
+
+    /**
+     * @brief Construct a LoanableSequence with a specified dynamic type.
+     *
+     * @param[in] dyn_type Pointer to the DynamicType.
+     */
+    LoanableSequence(
+            fastrtps::types::DynamicType_ptr dyn_type)
+        : dynamic_type_support_(new fastrtps::types::DynamicPubSubType(dyn_type))
+    {
+    }
+
+    /**
+     * @brief Construct a LoanableSequence with a specified maximum size.
+     *
+     * @param[in] max Maximum size of the sequence.
+     */
+    LoanableSequence(
+            size_type max)
+    {
+        if (max <= 0)
+        {
+            return;
+        }
+
+        resize(max);
+    }
+
+    /**
+     * @brief Destructor for LoanableSequence.
+     */
+    ~LoanableSequence()
+    {
+        if (elements_ && !has_ownership_)
+        {
+            EPROSIMA_LOG_WARNING(SUBSCRIBER, "Sequence destroyed with active loan");
+            return;
+        }
+
+        release();
+    }
+
+    /// Deleted copy constructor for LoanableSequence.
+    LoanableSequence(
+            const LoanableSequence& other) = delete;
+
+    /// Deleted copy assignment operator for LoanableSequence.
+    LoanableSequence& operator =(
+            const LoanableSequence& other) = delete;
+
+    /**
+     * @brief Move constructor for LoanableSequence.
+     *
+     * @param[in] other The other LoanableSequence to move from.
+     */
+    LoanableSequence(
+            LoanableSequence&&) = default;
+
+    /**
+     * @brief Move assignment operator for LoanableSequence.
+     *
+     * @param[in] other The other LoanableSequence to move from.
+     *
+     * @return A reference to this LoanableSequence.
+     */
+    LoanableSequence& operator =(
+            LoanableSequence&&) = default;
+
+protected:
+
+    using LoanableCollection::maximum_;
+    using LoanableCollection::length_;
+    using LoanableCollection::elements_;
+    using LoanableCollection::has_ownership_;
+
+private:
+
+    /**
+     * @brief Resize the sequence to a new maximum size.
+     *
+     * @param[in] maximum The new maximum size.
+     */
+    void resize(
+            size_type maximum) override
+    {
+        assert(has_ownership_);
+
+        // Resize collection and get new pointer
+        data_.reserve(maximum);
+        data_.resize(maximum);
+        elements_ = reinterpret_cast<element_type*>(data_.data());
+
+        // Allocate individual elements
+        while (maximum_ < maximum)
+        {
+            data_[maximum_++] = static_cast<fastrtps::types::DynamicData*>(dynamic_type_support_->createData());
+        }
+    }
+
+    /**
+     * @brief Release all elements and clear the sequence.
+     */
+    void release()
+    {
+        if (has_ownership_ && elements_)
+        {
+            for (size_type n = 0; n < maximum_; ++n)
+            {
+                fastrtps::types::DynamicData* elem = data_[n];
+                dynamic_type_support_->deleteData(elem);
+            }
+            std::vector<fastrtps::types::DynamicData*>().swap(data_);
+        }
+
+        maximum_ = 0u;
+        length_ = 0u;
+        elements_ = nullptr;
+        has_ownership_ = true;
+    }
+
+    /// Container for holding the DynamicData elements.
+    std::vector<fastrtps::types::DynamicData*> data_;
+
+    /// Pointer to the DynamicPubSubType type support.
+    std::unique_ptr<fastrtps::types::DynamicPubSubType> dynamic_type_support_;
+};
+
+/// Alias for LoanableSequence with DynamicData and true_type.
+using DynamicLoanableSequence = LoanableSequence<fastrtps::types::DynamicData, std::true_type>;
+
+} // namespace dds
+} // namespace fastdds
+} // namespace eprosima
+
+#endif // _TYPES_DYNAMIC_LOANABLE_SEQUENCE_HPP_
diff --git a/include/fastrtps/types/DynamicPubSubType.h b/include/fastrtps/types/DynamicPubSubType.h
@@ -15,10 +15,10 @@
 #ifndef TYPES_DYNAMIC_PUB_SUB_TYPE_H
 #define TYPES_DYNAMIC_PUB_SUB_TYPE_H
 
-#include <fastrtps/types/TypesBase.h>
 #include <fastdds/dds/topic/TopicDataType.hpp>
+#include <fastrtps/types/DynamicData.h>
 #include <fastrtps/types/DynamicTypePtr.h>
-#include <fastrtps/types/DynamicDataPtr.h>
+#include <fastrtps/types/TypesBase.h>
 #include <fastrtps/utils/md5.h>
 
 namespace eprosima {
@@ -45,6 +45,8 @@ class DynamicPubSubType : public eprosima::fastdds::dds::TopicDataType
 
 public:
 
+    typedef DynamicData type;
+
     RTPS_DllAPI DynamicPubSubType();
 
     RTPS_DllAPI DynamicPubSubType(

diff --git a/src/cpp/CMakeLists.txt b/src/cpp/CMakeLists.txt
@@ -65,6 +65,7 @@ set(${PROJECT_NAME}_source_files
     rtps/DataSharing/DataSharingListener.cpp
     rtps/DataSharing/DataSharingNotification.cpp
     rtps/reader/WriterProxy.cpp
+    rtps/reader/reader_utils.cpp
     rtps/reader/StatefulReader.cpp
     rtps/reader/StatelessReader.cpp
     rtps/reader/RTPSReader.cpp

diff --git a/src/cpp/fastdds/publisher/filtering/ReaderFilterCollection.hpp b/src/cpp/fastdds/publisher/filtering/ReaderFilterCollection.hpp
@@ -133,13 +133,17 @@ class ReaderFilterCollection
                         // Copy the signature
                         std::copy(entry.filter_signature.begin(), entry.filter_signature.end(), signature);
 
-                        // Evaluate filter and update filtered_out_readers
-                        bool filter_result = entry.filter->evaluate(change.serializedPayload, info, it->first);
-                        if (!filter_result)
+                        // Only evaluate filter on ALIVE changes, as UNREGISTERED and DISPOSED are always relevant
+                        bool filter_result = true;
+                        if (fastrtps::rtps::ALIVE == change.kind)
                         {
-                            change.filtered_out_readers.emplace_back(it->first);
+                            // Evaluate filter and update filtered_out_readers
+                            filter_result = entry.filter->evaluate(change.serializedPayload, info, it->first);
+                            if (!filter_result)
+                            {
+                                change.filtered_out_readers.emplace_back(it->first);
+                            }
                         }
-
                         return filter_result;
                     };
 

diff --git a/src/cpp/rtps/reader/StatefulReader.cpp b/src/cpp/rtps/reader/StatefulReader.cpp
@@ -32,6 +32,7 @@
 #include <fastdds/rtps/writer/LivelinessManager.h>
 #include <fastrtps/utils/TimeConversion.h>
 
+#include "reader_utils.hpp"
 #include <rtps/DataSharing/DataSharingListener.hpp>
 #include <rtps/DataSharing/ReaderPool.hpp>
 #include <rtps/history/HistoryAttributesExtension.hpp>
@@ -596,14 +597,15 @@ bool StatefulReader::processDataMsg(
                 return false;
             }
 
-            if (data_filter_ && !data_filter_->is_relevant(*change, m_guid))
+            if (!fastdds::rtps::change_is_relevant_for_filter(*change, m_guid, data_filter_))
             {
                 if (pWP)
                 {
                     pWP->irrelevant_change_set(change->sequenceNumber);
                     NotifyChanges(pWP);
                     send_ack_if_datasharing(this, mp_history, pWP, change->sequenceNumber);
                 }
+                // Change was filtered out, so there isn't anything else to do
                 return true;
             }
 
@@ -779,7 +781,8 @@ bool StatefulReader::processDataFragMsg(
 
                     // Temporarilly assign the inline qos while evaluating the data filter
                     work_change->inline_qos = incomingChange->inline_qos;
-                    bool filtered_out = data_filter_ && !data_filter_->is_relevant(*work_change, m_guid);
+                    bool filtered_out =
+                            !fastdds::rtps::change_is_relevant_for_filter(*work_change, m_guid, data_filter_);
                     work_change->inline_qos = SerializedPayload_t();
 
                     if (filtered_out)

diff --git a/src/cpp/rtps/reader/StatelessReader.cpp b/src/cpp/rtps/reader/StatelessReader.cpp
@@ -31,11 +31,11 @@
 #include <fastdds/rtps/reader/StatelessReader.h>
 #include <fastdds/rtps/writer/LivelinessManager.h>
 
-#include <rtps/participant/RTPSParticipantImpl.h>
+#include "reader_utils.hpp"
+#include "rtps/RTPSDomainImpl.hpp"
 #include <rtps/DataSharing/DataSharingListener.hpp>
 #include <rtps/DataSharing/ReaderPool.hpp>
-
-#include "rtps/RTPSDomainImpl.hpp"
+#include <rtps/participant/RTPSParticipantImpl.h>
 
 #define IDSTRING "(ID:" << std::this_thread::get_id() << ") " <<
 
@@ -589,9 +589,10 @@ bool StatelessReader::processDataMsg(
                 return false;
             }
 
-            if (data_filter_ && !data_filter_->is_relevant(*change, m_guid))
+            if (!fastdds::rtps::change_is_relevant_for_filter(*change, m_guid, data_filter_))
             {
                 update_last_notified(change->writerGUID, change->sequenceNumber);
+                // Change was filtered out, so there isn't anything else to do
                 return true;
             }
 
@@ -804,7 +805,8 @@ bool StatelessReader::processDataFragMsg(
                 {
                     // Temporarilly assign the inline qos while evaluating the data filter
                     change_completed->inline_qos = incomingChange->inline_qos;
-                    bool filtered_out = data_filter_ && !data_filter_->is_relevant(*change_completed, m_guid);
+                    bool filtered_out = !fastdds::rtps::change_is_relevant_for_filter(*change_completed, m_guid,
+                                    data_filter_);
                     change_completed->inline_qos = SerializedPayload_t();
 
                     if (filtered_out)

diff --git a/src/cpp/rtps/reader/reader_utils.cpp b/src/cpp/rtps/reader/reader_utils.cpp
@@ -0,0 +1,45 @@
+// 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 reader_utils.cpp
+ */
+
+#include "reader_utils.hpp"
+
+#include <fastdds/rtps/common/ChangeKind_t.hpp>
+
+namespace eprosima {
+namespace fastdds {
+namespace rtps {
+
+bool change_is_relevant_for_filter(
+        const CacheChange& change,
+        const GUID& reader_guid,
+        const IReaderDataFilter* filter)
+{
+    bool ret = true;
+
+    // Only evaluate filter on ALIVE changes, as UNREGISTERED and DISPOSED are always relevant
+    if ((nullptr != filter) && (fastrtps::rtps::ALIVE == change.kind) && (!filter->is_relevant(change, reader_guid)))
+    {
+        ret = false;
+    }
+
+    return ret;
+}
+
+} // namespace rtps
+} // namespace fastdds
+} // namespace eprosima