New FlowController [11831]

cmake/common/gtest.cmake

@@ -56,13 +56,15 @@ macro(add_gtest)
         endif()
 
         foreach(GTEST_SOURCE_FILE ${GTEST_SOURCES})
-            file(STRINGS ${GTEST_SOURCE_FILE} GTEST_TEST_NAMES REGEX ^TEST)
+            # Normal tests
+            file(STRINGS ${GTEST_SOURCE_FILE} GTEST_TEST_NAMES REGEX "^([T][Y][P][E][D][_])?TEST")
             foreach(GTEST_TEST_NAME ${GTEST_TEST_NAMES})
                 string(REGEX REPLACE ["\) \(,"] ";" GTEST_TEST_NAME ${GTEST_TEST_NAME})
                 list(GET GTEST_TEST_NAME 1 GTEST_GROUP_NAME)
                 list(GET GTEST_TEST_NAME 3 GTEST_TEST_NAME)
                 add_test(NAME ${GTEST_GROUP_NAME}.${GTEST_TEST_NAME}
-                    COMMAND ${command} --gtest_filter=${GTEST_GROUP_NAME}.${GTEST_TEST_NAME}:*/${GTEST_GROUP_NAME}.${GTEST_TEST_NAME}/*)
+                    COMMAND ${command}
+                    --gtest_filter=${GTEST_GROUP_NAME}.${GTEST_TEST_NAME}:*/${GTEST_GROUP_NAME}.${GTEST_TEST_NAME}/*:${GTEST_GROUP_NAME}/*.${GTEST_TEST_NAME})
 
                 # Add environment
                 set(GTEST_ENVIRONMENT "")
@@ -84,6 +86,7 @@ macro(add_gtest)
                 set_property(TEST ${GTEST_GROUP_NAME}.${GTEST_TEST_NAME} PROPERTY LABELS "${GTEST_LABELS}")
 
             endforeach()
+
         endforeach()
     else()
         add_test(NAME ${test} COMMAND ${command})

examples/C++/RTPSTest_as_socket/TestWriterSocket.cpp

@@ -92,16 +92,17 @@ void TestWriterSocket::run(
 {
     for (int i = 0; i < nmsgs; ++i )
     {
-        CacheChange_t* ch = mp_writer->new_change([]() -> uint32_t {
-            return 255;
-        }, ALIVE);
+        CacheChange_t* ch = mp_writer->new_change([]() -> uint32_t
+                        {
+                            return 255;
+                        }, ALIVE);
 #if defined(_WIN32)
         ch->serializedPayload.length =
                 sprintf_s((char*)ch->serializedPayload.data, 255, "My example string %d", i) + 1;
 #else
         ch->serializedPayload.length =
                 sprintf((char*)ch->serializedPayload.data, "My example string %d", i) + 1;
-#endif
+#endif // if defined(_WIN32)
         printf("Sending: %s\n", (char*)ch->serializedPayload.data);
         mp_history->add_change(ch);
     }

include/fastdds/rtps/attributes/RTPSParticipantAttributes.h

@@ -29,6 +29,7 @@
 #include <fastrtps/utils/fixed_size_string.hpp>
 #include <fastdds/rtps/attributes/RTPSParticipantAllocationAttributes.hpp>
 #include <fastdds/rtps/attributes/ServerAttributes.h>
+#include <fastdds/rtps/flowcontrol/FlowControllerDescriptor.hpp>
 
 #include <memory>
 #include <sstream>
@@ -438,6 +439,8 @@ class BuiltinAttributes
  */
 class RTPSParticipantAttributes
 {
+    using FlowControllerDescriptorList = std::vector<std::shared_ptr<fastdds::rtps::FlowControllerDescriptor>>;
+
 public:
 
     RTPSParticipantAttributes()
@@ -467,8 +470,9 @@ class RTPSParticipantAttributes
                (this->participantID == b.participantID) &&
                (this->throughputController == b.throughputController) &&
                (this->useBuiltinTransports == b.useBuiltinTransports) &&
-               (this->properties == b.properties &&
-               (this->prefix == b.prefix));
+               (this->properties == b.properties) &&
+               (this->prefix == b.prefix) &&
+               (this->flow_controllers == b.flow_controllers);
     }
 
     /**
@@ -542,6 +546,9 @@ class RTPSParticipantAttributes
         return name.c_str();
     }
 
+    //! Flow controllers.
+    FlowControllerDescriptorList flow_controllers;
+
 private:
 
     //!Name of the participant.

include/fastdds/rtps/attributes/WriterAttributes.h

@@ -23,6 +23,7 @@
 #include <fastdds/rtps/common/Guid.h>
 #include <fastdds/rtps/flowcontrol/ThroughputControllerDescriptor.h>
 #include <fastdds/rtps/attributes/EndpointAttributes.h>
+#include <fastdds/rtps/flowcontrol/FlowControllerConsts.hpp>
 #include <fastrtps/utils/collections/ResourceLimitedContainerConfig.hpp>
 #include <fastrtps/qos/QosPolicies.h>
 
@@ -138,6 +139,9 @@ class WriterAttributes
 
     //! Keep duration to keep a sample before considering it has been acked
     Duration_t keep_duration;
+
+    //! Flow controller name. Default: fastdds::rtps::FASTDDS_FLOW_CONTROLLER_DEFAULT.
+    const char* flow_controller_name = fastdds::rtps::FASTDDS_FLOW_CONTROLLER_DEFAULT;
 };
 
 } /* namespace rtps */

include/fastdds/rtps/common/CacheChange.h

@@ -35,6 +35,30 @@ namespace eprosima {
 namespace fastrtps {
 namespace rtps {
 
+/*!
+ * Specific information for a writer.
+ */
+struct CacheChangeWriterInfo_t
+{
+    //!Number of DATA / DATA_FRAG submessages sent to the transport (only used in Writers)
+    size_t num_sent_submessages = 0;
+    //! Used to link with previous node in a list. Used by FlowControllerImpl.
+    //! Cannot be cached because there are several comparisons without locking.
+    CacheChange_t* volatile previous = nullptr;
+    //! Used to link with next node in a list. Used by FlowControllerImpl.
+    //! Cannot be cached because there are several comparisons without locking.
+    CacheChange_t* volatile next = nullptr;
+};
+
+/*!
+ * Specific information for a reader.
+ */
+struct CacheChangeReaderInfo_t
+{
+    //!Reception TimeStamp (only used in Readers)
+    Time_t receptionTimestamp;
+};
+
 /**
  * Structure CacheChange_t, contains information on a specific CacheChange.
  * @ingroup COMMON_MODULE
@@ -57,10 +81,8 @@ struct RTPS_DllAPI CacheChange_t
     Time_t sourceTimestamp{};
     union
     {
-        //!Reception TimeStamp (only used in Readers)
-        Time_t receptionTimestamp;
-        //!Number of DATA / DATA_FRAG submessages sent to the transport (only used in Writers)
-        size_t num_sent_submessages = 0;
+        CacheChangeReaderInfo_t reader_info;
+        CacheChangeWriterInfo_t writer_info;
     };
 
     WriteParams write_params{};
@@ -105,7 +127,7 @@ struct RTPS_DllAPI CacheChange_t
         instanceHandle = ch_ptr->instanceHandle;
         sequenceNumber = ch_ptr->sequenceNumber;
         sourceTimestamp = ch_ptr->sourceTimestamp;
-        receptionTimestamp = ch_ptr->receptionTimestamp;
+        reader_info.receptionTimestamp = ch_ptr->reader_info.receptionTimestamp;
         write_params = ch_ptr->write_params;
         isRead = ch_ptr->isRead;
         fragment_size_ = ch_ptr->fragment_size_;
@@ -128,7 +150,7 @@ struct RTPS_DllAPI CacheChange_t
         instanceHandle = ch_ptr->instanceHandle;
         sequenceNumber = ch_ptr->sequenceNumber;
         sourceTimestamp = ch_ptr->sourceTimestamp;
-        receptionTimestamp = ch_ptr->receptionTimestamp;
+        reader_info.receptionTimestamp = ch_ptr->reader_info.receptionTimestamp;
         write_params = ch_ptr->write_params;
         isRead = ch_ptr->isRead;
 

include/fastdds/rtps/flowcontrol/FlowControllerSchedulerPolicy.hpp

@@ -31,8 +31,8 @@ enum class FlowControllerSchedulerPolicy : int32_t
 {
     //! FIFO scheduler policy: first written sample by user, first sample scheduled to be sent to network.
     FIFO,
-    //! Round Robind scheduler policy: schedules one sample of each DataWriter in circular order.
-    ROUND_ROBLIN,
+    //! Round Robin scheduler policy: schedules one sample of each DataWriter in circular order.
+    ROUND_ROBIN,
     //! High priority scheduler policy: samples with highest priority are scheduled first to be sent to network.
     HIGH_PRIORITY,
     //! Priority with reservation scheduler policy: guarantee each DataWriter's minimum reservation of throughput.

include/fastdds/rtps/history/WriterHistory.h

@@ -37,6 +37,7 @@ class WriterHistory : public rtps::History
 {
     friend class RTPSWriter;
     friend class PersistentWriter;
+    friend class IPersistenceService;
 
     WriterHistory(
             WriterHistory&&) = delete;
@@ -133,6 +134,13 @@ class WriterHistory : public rtps::History
     SequenceNumber_t m_lastCacheChangeSeqNum;
     //!Pointer to the associated RTPSWriter;
     RTPSWriter* mp_writer;
+
+    uint32_t high_mark_for_frag_ = 0;
+
+private:
+
+    void set_fragments(
+            CacheChange_t* change);
 };
 
 } // namespace rtps