Refs #22506. Filter interested readers on PDP writer

Signed-off-by: Eugenio Collado <[email protected]>

Author: EugenioCollado

include/fastdds/rtps/writer/StatelessWriter.h

@@ -76,6 +76,12 @@ class StatelessWriter : public RTPSWriter
             WriterHistory* hist,
             WriterListener* listen = nullptr);
 
+    mutable LocatorList_t fixed_locators_;
+
+    virtual bool send_to_fixed_locators(
+            CDRMessage_t* message,
+            std::chrono::steady_clock::time_point& max_blocking_time_point) const;
+
 public:
 
     virtual ~StatelessWriter();
@@ -159,10 +165,11 @@ class StatelessWriter : public RTPSWriter
         //FOR NOW THERE IS NOTHING TO UPDATE.
     }
 
+    //! Deprecated in favor of PDP simple writer
     bool set_fixed_locators(
             const LocatorList_t& locator_list);
 
-    //!Reset the unsent changes.
+    //! Deprecated in favor of PDP simple writer
     void unsent_changes_reset();
 
     /**
@@ -272,7 +279,6 @@ class StatelessWriter : public RTPSWriter
 
 
     bool is_inline_qos_expected_ = false;
-    LocatorList_t fixed_locators_;
     ResourceLimitedVector<std::unique_ptr<ReaderLocator>> matched_remote_readers_;
 
     std::condition_variable_any unsent_changes_cond_;

src/cpp/CMakeLists.txt

@@ -195,6 +195,7 @@ set(${PROJECT_NAME}_source_files
     rtps/builtin/discovery/participant/PDP.cpp
     rtps/builtin/discovery/participant/ServerAttributes.cpp
     rtps/builtin/discovery/participant/PDPSimple.cpp
+    rtps/builtin/discovery/participant/simple/PDPStatelessWriter.cpp
     rtps/builtin/discovery/participant/PDPListener.cpp
     rtps/builtin/discovery/endpoint/EDP.cpp
     rtps/builtin/discovery/endpoint/EDPSimple.cpp

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

@@ -285,7 +285,7 @@ void PDPSimple::announceParticipantState(
 
         if (!(dispose || new_change))
         {
-            endpoints->writer.writer_->unsent_changes_reset();
+            endpoints->writer.writer_->send_periodic_announcement();
         }
     }
 }
@@ -399,7 +399,7 @@ bool PDPSimple::create_dcps_participant_endpoints()
     if (mp_RTPSParticipant->createWriter(&rtps_writer, watt, writer.payload_pool_, writer.history_.get(),
             nullptr, writer_entity_id, true))
     {
-        writer.writer_ = dynamic_cast<StatelessWriter*>(rtps_writer);
+        writer.writer_ = dynamic_cast<PDPStatelessWriter*>(rtps_writer);
         assert(nullptr != writer.writer_);
 
 #if HAVE_SECURITY
@@ -451,7 +451,7 @@ bool PDPSimple::create_dcps_participant_endpoints()
                 EPROSIMA_LOG_WARNING(RTPS_PDP, "Ignoring initial peers locator " << loc << " : not allowed.");
             }
         }
-        writer.writer_->set_fixed_locators(fixed_locators);
+        writer.writer_->set_initial_peers(fixed_locators);
     }
     else
     {
@@ -721,11 +721,6 @@ void PDPSimple::match_pdp_remote_endpoints(
         {
             writer->matched_reader_add(*temp_reader_data);
         }
-
-        if (!writer_only && (BEST_EFFORT_RELIABILITY_QOS == reliability_kind))
-        {
-            endpoints->writer.writer_->unsent_changes_reset();
-        }
     }
 }
 

src/cpp/rtps/builtin/discovery/participant/simple/PDPStatelessWriter.cpp

@@ -0,0 +1,179 @@
+// Copyright 2025 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 PDPStatelessWriter.cpp
+ */
+
+#include <rtps/builtin/discovery/participant/simple/PDPStatelessWriter.hpp>
+
+#include <algorithm>
+#include <cassert>
+#include <chrono>
+#include <cstdint>
+#include <mutex>
+#include <set>
+#include <vector>
+
+#include <fastrtps/rtps/history/WriterHistory.h>
+#include <rtps/participant/RTPSParticipantImpl.h>
+
+namespace eprosima {
+namespace fastrtps {
+namespace rtps {
+
+PDPStatelessWriter::PDPStatelessWriter(
+        RTPSParticipantImpl* participant,
+        const GUID_t& guid,
+        const WriterAttributes& attributes,
+        fastdds::rtps::FlowController* flow_controller,
+        WriterHistory* history,
+        WriterListener* listener)
+    : StatelessWriter(participant, guid, attributes, flow_controller, history, listener)
+    , interested_readers_(participant->getRTPSParticipantAttributes().allocation.participants)
+{
+}
+
+bool PDPStatelessWriter::matched_reader_add(
+        const ReaderProxyData& data)
+{
+    bool ret = StatelessWriter::matched_reader_add(data);
+    if (ret)
+    {
+        // Mark new reader as interested
+        add_interested_reader(data.guid());
+        // Send announcement to new reader
+        reschedule_all_samples();
+    }
+    return ret;
+}
+
+bool PDPStatelessWriter::matched_reader_remove(
+        const GUID_t& reader_guid)
+{
+    bool ret = StatelessWriter::matched_reader_remove(reader_guid);
+    if (ret)
+    {
+        // Mark reader as not interested
+        remove_interested_reader(reader_guid);
+    }
+    return ret;
+}
+
+void PDPStatelessWriter::unsent_change_added_to_history(
+        CacheChange_t* change,
+        const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time)
+{
+    mark_all_readers_interested();
+    StatelessWriter::unsent_change_added_to_history(change, max_blocking_time);
+}
+
+void PDPStatelessWriter::set_initial_peers(
+        const fastdds::rtps::LocatorList& locator_list)
+{
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+
+    initial_peers_.push_back(locator_list);
+    mp_RTPSParticipant->createSenderResources(initial_peers_);
+}
+
+void PDPStatelessWriter::send_periodic_announcement()
+{
+    mark_all_readers_interested();
+    reschedule_all_samples();
+}
+
+bool PDPStatelessWriter::send_to_fixed_locators(
+        CDRMessage_t* message,
+        std::chrono::steady_clock::time_point& max_blocking_time_point) const
+{
+    bool ret = true;
+
+    if (should_reach_all_destinations_)
+    {
+        ret = initial_peers_.empty() ||
+                mp_RTPSParticipant->sendSync(message, m_guid,
+                        Locators(initial_peers_.begin()), Locators(initial_peers_.end()),
+                        max_blocking_time_point);
+
+        if (ret)
+        {
+            fixed_locators_.clear();
+            should_reach_all_destinations_ = false;
+        }
+    }
+    else
+    {
+        interested_readers_.clear();
+    }
+
+    return ret;
+}
+
+bool PDPStatelessWriter::is_relevant(
+        const CacheChange_t& /* change */,
+        const GUID_t& reader_guid) const
+{
+    return interested_readers_.end() !=
+           std::find(interested_readers_.begin(), interested_readers_.end(), reader_guid);
+}
+
+void PDPStatelessWriter::mark_all_readers_interested()
+{
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    should_reach_all_destinations_ = true;
+    interested_readers_.clear();
+    fixed_locators_.clear();
+    fixed_locators_.push_back(initial_peers_);
+    reader_data_filter(nullptr);
+}
+
+void PDPStatelessWriter::add_interested_reader(
+        const GUID_t& reader_guid)
+{
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    if (!should_reach_all_destinations_)
+    {
+        auto it = std::find(interested_readers_.begin(), interested_readers_.end(), reader_guid);
+        if (it == interested_readers_.end())
+        {
+            interested_readers_.emplace_back(reader_guid);
+            reader_data_filter(this);
+        }
+    }
+}
+
+void PDPStatelessWriter::remove_interested_reader(
+        const GUID_t& reader_guid)
+{
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    interested_readers_.remove(reader_guid);
+}
+
+void PDPStatelessWriter::reschedule_all_samples()
+{
+    std::lock_guard<RecursiveTimedMutex> guard(mp_mutex);
+    size_t n_samples = mp_history->getHistorySize();
+    if (0 < n_samples)
+    {
+        assert(1 == n_samples);
+        auto it = mp_history->changesBegin();
+        CacheChange_t* change = *it;
+        flow_controller_->add_new_sample(this, change, std::chrono::steady_clock::now() + std::chrono::hours(24));
+    }
+}
+
+} // namespace rtps
+} // namespace fastdds
+} // namespace eprosima