Limit RTPS History Depth (#288)

juanlofer-eprosima · web-flow · commit 45f85348d5e1 · 2022-08-31T12:40:07.000+02:00
* Remove duplicated Forthcoming section

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;

* Limit RTPS History Depth

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;

* Uncrustify

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;

* Fix typo

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;

* Make RTPS Writers Synchronous

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;

Signed-off-by: Juan López Fernández &lt;juanlopez@eprosima.com&gt;
diff --git a/ddsrouter_core/include/ddsrouter_core/configuration/DDSRouterConfiguration.hpp b/ddsrouter_core/include/ddsrouter_core/configuration/DDSRouterConfiguration.hpp
@@ -53,7 +53,8 @@ struct DDSRouterConfiguration : public DDSRouterReloadConfiguration
             std::set<std::shared_ptr<types::FilterTopic>> blocklist,
             std::set<std::shared_ptr<types::RealTopic>> builtin_topics,
             std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations,
-            unsigned int number_of_threads);
+            unsigned int number_of_threads,
+            unsigned int max_history_depth);
 
     /////////////////////////
     // METHODS
@@ -73,6 +74,8 @@ struct DDSRouterConfiguration : public DDSRouterReloadConfiguration
 
     unsigned int number_of_threads = 12;
 
+    unsigned int max_history_depth = 5000;
+
 protected:
 
     static bool check_correct_configuration_object_(
diff --git a/ddsrouter_core/include/ddsrouter_core/configuration/participant/SimpleParticipantConfiguration.hpp b/ddsrouter_core/include/ddsrouter_core/configuration/participant/SimpleParticipantConfiguration.hpp
@@ -67,6 +67,8 @@ struct SimpleParticipantConfiguration : public ParticipantConfiguration
     /////////////////////////
 
     types::DomainId domain = types::DomainId(0u);
+
+    unsigned int max_history_depth = 5000;
 };
 
 } /* namespace configuration */
diff --git a/ddsrouter_core/src/cpp/configuration/DDSRouterConfiguration.cpp b/ddsrouter_core/src/cpp/configuration/DDSRouterConfiguration.cpp
@@ -40,10 +40,12 @@ DDSRouterConfiguration::DDSRouterConfiguration(
         std::set<std::shared_ptr<FilterTopic>> blocklist,
         std::set<std::shared_ptr<RealTopic>> builtin_topics,
         std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations,
-        unsigned int number_of_threads /* = default_number_of_threads() */)
+        unsigned int number_of_threads,
+        unsigned int max_history_depth)
     : DDSRouterReloadConfiguration (allowlist, blocklist, builtin_topics)
     , participants_configurations(participants_configurations)
     , number_of_threads(number_of_threads)
+    , max_history_depth(max_history_depth)
 {
 }
 
diff --git a/ddsrouter_core/src/cpp/core/DDSRouterImpl.cpp b/ddsrouter_core/src/cpp/core/DDSRouterImpl.cpp
@@ -288,7 +288,8 @@ void DDSRouterImpl::init_participants_()
                 participant_factory_.create_participant(
             participant_config,
             payload_pool_,
-            discovery_database_);
+            discovery_database_,
+            configuration_.max_history_depth);
 
         // create_participant should throw an exception in fail, never return nullptr
         if (!new_participant || !new_participant->id().is_valid() ||
diff --git a/ddsrouter_core/src/cpp/core/ParticipantFactory.cpp b/ddsrouter_core/src/cpp/core/ParticipantFactory.cpp
@@ -43,7 +43,8 @@ using namespace eprosima::ddsrouter::core::configuration;
 std::shared_ptr<IParticipant> ParticipantFactory::create_participant(
         std::shared_ptr<configuration::ParticipantConfiguration> participant_configuration,
         std::shared_ptr<PayloadPool> payload_pool,
-        std::shared_ptr<DiscoveryDatabase> discovery_database)
+        std::shared_ptr<DiscoveryDatabase> discovery_database,
+        unsigned int max_history_depth)
 {
     // Create a new Participant depending on the ParticipantKind specified by the configuration
     switch (participant_configuration->kind)
@@ -80,6 +81,8 @@ std::shared_ptr<IParticipant> ParticipantFactory::create_participant(
             std::shared_ptr<configuration::SimpleParticipantConfiguration> conf_ =
                     std::dynamic_pointer_cast<configuration::SimpleParticipantConfiguration>(
                 participant_configuration);
+            // TMP: Until Transparency QoS module is available
+            conf_->max_history_depth = max_history_depth;
             if (!conf_)
             {
                 throw utils::ConfigurationException(
@@ -100,6 +103,8 @@ std::shared_ptr<IParticipant> ParticipantFactory::create_participant(
             std::shared_ptr<configuration::DiscoveryServerParticipantConfiguration> conf_ =
                     std::dynamic_pointer_cast<configuration::DiscoveryServerParticipantConfiguration>(
                 participant_configuration);
+            // TMP: Until Transparency QoS module is available
+            conf_->max_history_depth = max_history_depth;
             if (!conf_)
             {
                 throw utils::ConfigurationException(
@@ -119,6 +124,8 @@ std::shared_ptr<IParticipant> ParticipantFactory::create_participant(
             std::shared_ptr<configuration::InitialPeersParticipantConfiguration> conf_ =
                     std::dynamic_pointer_cast<configuration::InitialPeersParticipantConfiguration>(
                 participant_configuration);
+            // TMP: Until Transparency QoS module is available
+            conf_->max_history_depth = max_history_depth;
             if (!conf_)
             {
                 throw utils::ConfigurationException(
diff --git a/ddsrouter_core/src/cpp/core/ParticipantFactory.hpp b/ddsrouter_core/src/cpp/core/ParticipantFactory.hpp
@@ -40,12 +40,14 @@ class ParticipantFactory
      *      participant that will be created, and its Configuration.
      * @param [in] payload : Common Payload pool to create the Participant
      * @param [in] discovery_database : Common Discovery Database to create the Participant
+     * @param [in] max_history_depth : Maximum size of RTPS History instances (only applies to RTPS participants);
      * @return new Participant
      */
     std::shared_ptr<IParticipant> create_participant(
             std::shared_ptr<configuration::ParticipantConfiguration> participant_configuration,
             std::shared_ptr<PayloadPool> payload,
-            std::shared_ptr<DiscoveryDatabase> discovery_database);
+            std::shared_ptr<DiscoveryDatabase> discovery_database,
+            unsigned int max_history_depth);
 
     /**
      * @brief Delete correctly a Participant
diff --git a/ddsrouter_core/src/cpp/participant/implementations/rtps/CommonParticipant.cpp b/ddsrouter_core/src/cpp/participant/implementations/rtps/CommonParticipant.cpp
@@ -41,8 +41,10 @@ CommonParticipant::CommonParticipant(
         std::shared_ptr<PayloadPool> payload_pool,
         std::shared_ptr<DiscoveryDatabase> discovery_database,
         const types::DomainId& domain_id,
-        const fastrtps::rtps::RTPSParticipantAttributes& participant_attributes)
+        const fastrtps::rtps::RTPSParticipantAttributes& participant_attributes,
+        unsigned int max_history_depth)
     : BaseParticipant(participant_configuration, payload_pool, discovery_database)
+    , max_history_depth_(max_history_depth)
 {
     create_participant_(
         domain_id,
@@ -242,6 +244,7 @@ std::shared_ptr<IWriter> CommonParticipant::create_writer_(
         topic,
         this->payload_pool_,
         rtps_participant_,
+        max_history_depth_,
         this->configuration_->is_repeater);
 }
 
diff --git a/ddsrouter_core/src/cpp/participant/implementations/rtps/CommonParticipant.hpp b/ddsrouter_core/src/cpp/participant/implementations/rtps/CommonParticipant.hpp
@@ -53,7 +53,8 @@ class CommonParticipant
             std::shared_ptr<PayloadPool> payload_pool,
             std::shared_ptr<DiscoveryDatabase> discovery_database,
             const types::DomainId& domain_id,
-            const fastrtps::rtps::RTPSParticipantAttributes& participant_attributes);
+            const fastrtps::rtps::RTPSParticipantAttributes& participant_attributes,
+            unsigned int max_history_depth);
 
     virtual ~CommonParticipant();
 
@@ -97,6 +98,9 @@ class CommonParticipant
 
     //! Mutex that guards every access to the RTPS Participant
     mutable std::recursive_mutex rtps_mutex_;
+
+    //! Maximum depth of RTPS History instances
+    unsigned int max_history_depth_;
 };
 
 } /* namespace rtps */
diff --git a/ddsrouter_core/src/cpp/participant/implementations/rtps/DiscoveryServerParticipant.cpp b/ddsrouter_core/src/cpp/participant/implementations/rtps/DiscoveryServerParticipant.cpp
@@ -42,7 +42,8 @@ DiscoveryServerParticipant::DiscoveryServerParticipant(
         payload_pool,
         discovery_database,
         participant_configuration->domain,
-        participant_attributes_(participant_configuration.get()))
+        participant_attributes_(participant_configuration.get()),
+        participant_configuration->max_history_depth)
 {
 }
 
diff --git a/ddsrouter_core/src/cpp/participant/implementations/rtps/InitialPeersParticipant.cpp b/ddsrouter_core/src/cpp/participant/implementations/rtps/InitialPeersParticipant.cpp
@@ -39,7 +39,8 @@ InitialPeersParticipant::InitialPeersParticipant(
         payload_pool,
         discovery_database,
         participant_configuration->domain,
-        participant_attributes_(participant_configuration.get()))
+        participant_attributes_(participant_configuration.get()),
+        participant_configuration->max_history_depth)
 {
 }
 
diff --git a/ddsrouter_core/src/cpp/participant/implementations/rtps/SimpleParticipant.cpp b/ddsrouter_core/src/cpp/participant/implementations/rtps/SimpleParticipant.cpp
@@ -39,7 +39,8 @@ SimpleParticipant::SimpleParticipant(
         payload_pool,
         discovery_database,
         participant_configuration->domain,
-        CommonParticipant::participant_attributes_(participant_configuration.get()))
+        CommonParticipant::participant_attributes_(participant_configuration.get()),
+        participant_configuration->max_history_depth)
 {
 }
 
diff --git a/ddsrouter_core/src/cpp/writer/implementations/rtps/Writer.cpp b/ddsrouter_core/src/cpp/writer/implementations/rtps/Writer.cpp
@@ -40,6 +40,7 @@ Writer::Writer(
         const RealTopic& topic,
         std::shared_ptr<PayloadPool> payload_pool,
         fastrtps::rtps::RTPSParticipant* rtps_participant,
+        unsigned int max_history_depth,
         const bool repeater /* = false */)
     : BaseWriter(participant_id, topic, payload_pool)
     , repeater_(repeater)
@@ -48,6 +49,7 @@ Writer::Writer(
 
     // Create History
     fastrtps::rtps::HistoryAttributes history_att = history_attributes_();
+    history_att.maximumReservedCaches = max_history_depth;
     rtps_history_ = new fastrtps::rtps::WriterHistory(history_att);
 
     // Create Writer
@@ -182,7 +184,11 @@ utils::ReturnCode Writer::write_(
     // Send data by adding it to Writer History
     rtps_history_->add_change(new_change);
 
-    // TODO: Data is never removed till destruction
+    // When max history size is reached, remove oldest cache change
+    if (rtps_history_->isFull())
+    {
+        rtps_history_->remove_min_change();
+    }
 
     return utils::ReturnCode::RETCODE_OK;
 }
@@ -200,7 +206,8 @@ fastrtps::rtps::WriterAttributes Writer::writer_attributes_() const noexcept
     fastrtps::rtps::WriterAttributes att;
     att.endpoint.durabilityKind = eprosima::fastrtps::rtps::DurabilityKind_t::TRANSIENT_LOCAL;
     att.endpoint.reliabilityKind = eprosima::fastrtps::rtps::ReliabilityKind_t::RELIABLE;
-    att.mode = fastrtps::rtps::RTPSWriterPublishMode::ASYNCHRONOUS_WRITER;
+    // Write synchronously to avoid removing a change before being sent, likely event when history depth is very small
+    att.mode = fastrtps::rtps::RTPSWriterPublishMode::SYNCHRONOUS_WRITER;
     if (topic_.topic_with_key())
     {
         att.endpoint.topicKind = eprosima::fastrtps::rtps::WITH_KEY;
diff --git a/ddsrouter_core/src/cpp/writer/implementations/rtps/Writer.hpp b/ddsrouter_core/src/cpp/writer/implementations/rtps/Writer.hpp
@@ -80,6 +80,7 @@ class Writer : public BaseWriter
             const types::RealTopic& topic,
             std::shared_ptr<PayloadPool> payload_pool,
             fastrtps::rtps::RTPSParticipant* rtps_participant,
+            unsigned int max_history_depth,
             const bool repeater = false);
 
     /**
diff --git a/ddsrouter_core/test/blackbox/ddsrouter_core/dds/WAN/DDSTestWAN.cpp b/ddsrouter_core/test/blackbox/ddsrouter_core/dds/WAN/DDSTestWAN.cpp
@@ -187,7 +187,8 @@ configuration::DDSRouterConfiguration router_configuration(
         blocklist,
         builtin_topics,
         participants_configurations,
-        1);
+        1,
+        100);
 }
 
 /**
diff --git a/ddsrouter_core/test/blackbox/ddsrouter_core/dds/local/DDSTestLocal.cpp b/ddsrouter_core/test/blackbox/ddsrouter_core/dds/local/DDSTestLocal.cpp
@@ -86,8 +86,8 @@ configuration::DDSRouterConfiguration dds_test_simple_configuration(
         blocklist,
         builtin_topics,
         participants_configurations,
-        1
-        );
+        1,
+        100);
 }
 
 /**
diff --git a/ddsrouter_core/test/blackbox/implementations/ImplementationsTest.cpp b/ddsrouter_core/test/blackbox/implementations/ImplementationsTest.cpp
@@ -40,6 +40,7 @@ namespace ddsrouter {
 namespace test {
 
 constexpr const unsigned int DEFAULT_THREAD_POOL_SIZE = 2;
+constexpr const unsigned int DEFAULT_MAX_HISTORY_DEPTH = 100;
 
 } /* namespace test */
 } /* namespace ddsrouter */
@@ -76,7 +77,8 @@ TEST(ImplementationsTest, pair_implementation)
             std::set<std::shared_ptr<FilterTopic>>(),
             std::set<std::shared_ptr<RealTopic>>(),
             participant_configurations,
-            test::DEFAULT_THREAD_POOL_SIZE);
+            test::DEFAULT_THREAD_POOL_SIZE,
+            test::DEFAULT_MAX_HISTORY_DEPTH);
 
         // Create DDSRouter entity
         DDSRouter router(configuration);
@@ -121,7 +123,8 @@ TEST(ImplementationsTest, pair_implementation_with_topic)
             std::set<std::shared_ptr<FilterTopic>>(),
             builtin_topics,
             participant_configurations,
-            test::DEFAULT_THREAD_POOL_SIZE);
+            test::DEFAULT_THREAD_POOL_SIZE,
+            test::DEFAULT_MAX_HISTORY_DEPTH);
 
         // Create DDSRouter entity
         DDSRouter router(configuration);
@@ -173,7 +176,8 @@ TEST(ImplementationsTest, all_implementations)
             std::set<std::shared_ptr<FilterTopic>>(),
             std::set<std::shared_ptr<RealTopic>>(),
             participant_configurations,
-            test::DEFAULT_THREAD_POOL_SIZE);
+            test::DEFAULT_THREAD_POOL_SIZE,
+            test::DEFAULT_MAX_HISTORY_DEPTH);
 
         // Create DDSRouter entity
         DDSRouter router(configuration);
@@ -209,7 +213,8 @@ TEST(ImplementationsTest, duplicated_ids)
             std::set<std::shared_ptr<FilterTopic>>(),
             std::set<std::shared_ptr<RealTopic>>(),
             participant_configurations,
-            test::DEFAULT_THREAD_POOL_SIZE);
+            test::DEFAULT_THREAD_POOL_SIZE,
+            test::DEFAULT_MAX_HISTORY_DEPTH);
 
         // Create DDSRouter entity
         ASSERT_THROW(DDSRouter router(configuration), eprosima::ddsrouter::utils::ConfigurationException) << kind;
diff --git a/ddsrouter_core/test/unittest/core/participant_factory/ParticipantFactoryTest.cpp b/ddsrouter_core/test/unittest/core/participant_factory/ParticipantFactoryTest.cpp
@@ -33,6 +33,8 @@ namespace eprosima {
 namespace ddsrouter {
 namespace test {
 
+constexpr const unsigned int DEFAULT_MAX_HISTORY_DEPTH = 100;
+
 /*
  * Generate all required objects for participant creation,
  * and then create a participant from an id string
@@ -47,7 +49,8 @@ std::shared_ptr<IParticipant> create_participant(
     std::shared_ptr<configuration::ParticipantConfiguration> participant_configuration =
             random_participant_configuration(kind);
 
-    return participant_factory.create_participant(participant_configuration, payload_pool, discovery_database);
+    return participant_factory.create_participant(participant_configuration, payload_pool, discovery_database,
+                   test::DEFAULT_MAX_HISTORY_DEPTH);
 }
 
 } /* namespace test */
diff --git a/ddsrouter_yaml/include/ddsrouter_yaml/yaml_configuration_tags.hpp b/ddsrouter_yaml/include/ddsrouter_yaml/yaml_configuration_tags.hpp
@@ -98,6 +98,7 @@ constexpr const char* DISCOVERY_SERVER_ID_ROS_TAG("ros-discovery-server"); //! T
 
 // Advance configuration
 constexpr const char* NUMBER_THREADS_TAG("threads"); //! Number of threads to configure the thread pool
+constexpr const char* MAX_HISTORY_DEPTH_TAG("max-depth"); //! Maximum size (number of stored cache changes) for RTPS History instances
 
 // Old versions tags
 constexpr const char* PARTICIPANT_KIND_TAG_V1("type"); //! Participant Kind
diff --git a/ddsrouter_yaml/src/cpp/YamlReader.cpp b/ddsrouter_yaml/src/cpp/YamlReader.cpp
@@ -946,6 +946,13 @@ void _fill_ddsrouter_configuration_latest(
         object.number_of_threads = YamlReader::get<unsigned int>(yml, NUMBER_THREADS_TAG, version);
     }
 
+    /////
+    // Get optional maximum history depth
+    if (YamlReader::is_tag_present(yml, MAX_HISTORY_DEPTH_TAG))
+    {
+        object.max_history_depth = YamlReader::get<unsigned int>(yml, MAX_HISTORY_DEPTH_TAG, version);
+    }
+
 }
 
 template <>
diff --git a/ddsrouter_yaml/test/unittest/configuration/CMakeLists.txt b/ddsrouter_yaml/test/unittest/configuration/CMakeLists.txt
@@ -33,6 +33,7 @@ set(TEST_LIST
         get_ddsrouter_configuration_no_version
         version_negative_cases
         number_of_threads
+        max_history_depth
     )
 
 set(TEST_EXTRA_LIBRARIES
diff --git a/ddsrouter_yaml/test/unittest/configuration/YamlReaderConfigurationTest.cpp b/ddsrouter_yaml/test/unittest/configuration/YamlReaderConfigurationTest.cpp
@@ -278,6 +278,41 @@ TEST(YamlReaderConfigurationTest, number_of_threads)
     }
 }
 
+/**
+ * Test load of maximum history depth in the configuration
+ *
+ * CASES:
+ * - trivial configuration
+ */
+TEST(YamlReaderConfigurationTest, max_history_depth)
+{
+    const char* yml_configuration =
+            // trivial configuration
+            R"(
+        version: v2.0
+        participants:
+          - name: "P1"
+            kind: "void"
+          - name: "P2"
+            kind: "void"
+        )";
+    Yaml yml = YAML::Load(yml_configuration);
+
+    std::vector<unsigned int> test_cases = {10, 100, 1000, 5000, 10000};
+
+    for (unsigned int test_case : test_cases)
+    {
+        yml[MAX_HISTORY_DEPTH_TAG] = test_case;
+
+        // Load configuration
+        core::configuration::DDSRouterConfiguration configuration_result =
+                YamlReaderConfiguration::load_ddsrouter_configuration(yml);
+
+        // Check max history depth is correct
+        ASSERT_EQ(test_case, configuration_result.max_history_depth);
+    }
+}
+
 int main(
         int argc,
         char** argv)
diff --git a/docs/rst/notes/notes.rst b/docs/rst/notes/notes.rst
@@ -6,8 +6,6 @@
 
 .. include:: forthcoming_version.rst
 
-.. include:: forthcoming_version.rst
-
 ##############
 Version v0.4.0
 ##############
diff --git a/docs/rst/user_manual/configuration.rst b/docs/rst/user_manual/configuration.rst

Original file line number	Diff line number	Diff line change
`@@ -40,10 +40,12 @@ DDSRouterConfiguration::DDSRouterConfiguration(`
`40`	`40`	`std::set<std::shared_ptr<FilterTopic>> blocklist,`
`41`	`41`	`std::set<std::shared_ptr<RealTopic>> builtin_topics,`
`42`	`42`	`std::set<std::shared_ptr<ParticipantConfiguration>> participants_configurations,`
`43`		`- unsigned int number_of_threads /* = default_number_of_threads() */)`
	`43`	`+ unsigned int number_of_threads,`
	`44`	`+ unsigned int max_history_depth)`
`44`	`45`	`: DDSRouterReloadConfiguration (allowlist, blocklist, builtin_topics)`
`45`	`46`	`, participants_configurations(participants_configurations)`
`46`	`47`	`, number_of_threads(number_of_threads)`
	`48`	`+ , max_history_depth(max_history_depth)`
`47`	`49`	`{`
`48`	`50`	`}`
`49`	`51`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,8 @@ DiscoveryServerParticipant::DiscoveryServerParticipant(`
`42`	`42`	`payload_pool,`
`43`	`43`	`discovery_database,`
`44`	`44`	`participant_configuration->domain,`
`45`		`- participant_attributes_(participant_configuration.get()))`
	`45`	`+ participant_attributes_(participant_configuration.get()),`
	`46`	`+ participant_configuration->max_history_depth)`
`46`	`47`	`{`
`47`	`48`	`}`
`48`	`49`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,8 @@ InitialPeersParticipant::InitialPeersParticipant(`
`39`	`39`	`payload_pool,`
`40`	`40`	`discovery_database,`
`41`	`41`	`participant_configuration->domain,`
`42`		`- participant_attributes_(participant_configuration.get()))`
	`42`	`+ participant_attributes_(participant_configuration.get()),`
	`43`	`+ participant_configuration->max_history_depth)`
`43`	`44`	`{`
`44`	`45`	`}`
`45`	`46`