Remove out the responsibility for bandwidth limitation from RTPSMessageGroup (#6214)

* Refs #23867. Support several flowcontroller factories

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

* Refs #23867. Add new IRTPSMessageGroupLimitation

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

* Refs #23867. Fixes

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

* Refs #23867. Add more info to IRTPSMessageGroupLimitation

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

* Refs #23867. Add doxygen documentation

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

* Refs #23867. Apply suggestions

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

* Refs #23867. Little refactor

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

* Refs #23867. fix tests

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

* Refs #23867. Fix compilation error on Windows

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

---------

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

Author: richiware

src/cpp/fastdds/types.hpp

@@ -0,0 +1,14 @@
+#ifndef FASTDDS_TYPES_HPP
+#define FASTDDS_TYPES_HPP
+
+#include <rtps/flowcontrol/FlowControllerFactory.hpp>
+
+namespace eprosima {
+namespace fastdds {
+
+using FlowControllerFactoryType = fastdds::rtps::FlowControllerFactory;
+
+} // namespace fastdds
+} // namespace eprosima
+
+#endif // FASTDDS_TYPES_HPP

src/cpp/rtps/flowcontrol/FlowControllerFactory.hpp

@@ -38,7 +38,7 @@ class FlowControllerFactory
      *
      * @param participant Pointer to the participant owner of this object.
      */
-    void init(
+    virtual void init(
             fastdds::rtps::RTPSParticipantImpl* participant);
 
     /*!
@@ -60,7 +60,7 @@ class FlowControllerFactory
             const std::string& flow_controller_name,
             const fastdds::rtps::WriterAttributes& writer_attributes);
 
-private:
+protected:
 
     fastdds::rtps::RTPSParticipantImpl* participant_ = nullptr;
 

src/cpp/rtps/flowcontrol/FlowControllerImpl.hpp

@@ -13,6 +13,7 @@
 #include <fastdds/utils/TimedConditionVariable.hpp>
 #include <fastdds/utils/TimedMutex.hpp>
 
+#include <rtps/messages/IRTPSMessageGroupLimitation.hpp>
 #include <rtps/messages/RTPSMessageGroup.hpp>
 #include <rtps/participant/RTPSParticipantImpl.hpp>
 #include <rtps/writer/BaseWriter.hpp>
@@ -277,19 +278,20 @@ struct FlowControllerSyncPublishMode : public FlowControllerPureSyncPublishMode,
 };
 
 //! Sends all samples asynchronously but with bandwidth limitation.
-struct FlowControllerLimitedAsyncPublishMode : public FlowControllerAsyncPublishMode
+struct FlowControllerLimitedAsyncPublishMode : public FlowControllerAsyncPublishMode, public IRTPSMessageGroupLimitation
 {
     FlowControllerLimitedAsyncPublishMode(
             RTPSParticipantImpl* participant,
             const FlowControllerDescriptor* descriptor)
         : FlowControllerAsyncPublishMode(participant, descriptor)
+        , sent_bytes_limitation_(static_cast<uint32_t>(descriptor->max_bytes_per_period))
     {
         assert(nullptr != descriptor);
         assert(0 < descriptor->max_bytes_per_period);
 
         max_bytes_per_period = descriptor->max_bytes_per_period;
         period_ms = std::chrono::milliseconds(descriptor->period_ms);
-        group.set_sent_bytes_limitation(static_cast<uint32_t>(max_bytes_per_period));
+        group.set_limitation(this);
     }
 
     bool fast_check_is_there_slot_for_change(
@@ -311,7 +313,7 @@ struct FlowControllerLimitedAsyncPublishMode : public FlowControllerAsyncPublish
 
         }
 
-        bool ret = (max_bytes_per_period - group.get_current_bytes_processed()) > size_to_check;
+        bool ret = (max_bytes_per_period - current_sent_bytes_) > size_to_check;
 
         if (!ret)
         {
@@ -345,7 +347,7 @@ struct FlowControllerLimitedAsyncPublishMode : public FlowControllerAsyncPublish
         {
             last_period_ = std::chrono::steady_clock::now();
             force_wait_ = false;
-            group.reset_current_bytes_processed();
+            current_sent_bytes_ = 0;
         }
 
         return reset_limit;
@@ -365,15 +367,39 @@ struct FlowControllerLimitedAsyncPublishMode : public FlowControllerAsyncPublish
         }
     }
 
+    void add_sent_bytes_by_group(
+            uint32_t bytes,
+            RTPSMessageSenderInterface&) override
+    {
+        current_sent_bytes_ += bytes;
+    }
+
+    bool data_exceeds_limitation(
+            CacheChange_t&,
+            uint32_t size_to_add,
+            uint32_t pending_to_send,
+            RTPSMessageSenderInterface&) override
+    {
+        return
+            //   either limitation has already been reached
+            (sent_bytes_limitation_ <= (current_sent_bytes_ + pending_to_send)) ||
+            //   or adding size_to_add will exceed limitation
+            (size_to_add > (sent_bytes_limitation_ - (current_sent_bytes_ + pending_to_send)));
+    }
+
     int32_t max_bytes_per_period = 0;
 
     std::chrono::milliseconds period_ms;
 
 private:
 
-    bool force_wait_ = false;
+    bool force_wait_ {false};
+
+    std::chrono::steady_clock::time_point last_period_ {std::chrono::steady_clock::now()};
 
-    std::chrono::steady_clock::time_point last_period_ = std::chrono::steady_clock::now();
+    uint32_t sent_bytes_limitation_ {0};
+
+    uint32_t current_sent_bytes_ {0};
 };
 
 
@@ -1068,7 +1094,7 @@ class FlowControllerImpl : public FlowController
         return get_max_payload_impl();
     }
 
-private:
+protected:
 
     /*!
      * Initialize asynchronous thread.

src/cpp/rtps/messages/IRTPSMessageGroupLimitation.hpp

@@ -0,0 +1,73 @@
+// 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 IRTPSMessageGroupLimitation.h
+ *
+ */
+
+#ifndef FASTDDS_RTPS_MESSAGES__IRTPSMESSAGEGROUPLIMITATION_HPP
+#define FASTDDS_RTPS_MESSAGES__IRTPSMESSAGEGROUPLIMITATION_HPP
+
+#include <cstdint>
+
+namespace eprosima {
+namespace fastdds {
+namespace rtps {
+
+struct CacheChange_t;
+class RTPSMessageSenderInterface;
+
+/*!
+ * Interface to implement mechanisms that limit the data sent by an RTPSMessageGroup under certain conditions,
+ * taking into account the bytes already sent by said group.
+ */
+class IRTPSMessageGroupLimitation
+{
+public:
+
+    /*!
+     * RTPSMessageGroup uses this method to announce the number of bytes just sent by the group.
+     *
+     * @param bytes Number of bytes just sent by the group.
+     * @param sender RTPSMessageSenderInterface used for delivering sent bytes.
+     */
+    virtual void add_sent_bytes_by_group(
+            uint32_t bytes,
+            RTPSMessageSenderInterface& sender) = 0;
+
+    /*!
+     * RTPSMessageGroup uses this method to query whether adding a new change to the group would exceed the
+     * limitation or not.
+     *
+     * @param change Change to be added.
+     * @param size_to_add Size in bytes that adding the change would imply.
+     * @param pending_to_send Number of bytes pending to be sent by the group.
+     * @param sender RTPSMessageSenderInterface that the group will use for data delivery.
+     *
+     * @return True if adding the change would exceed the limitation, false otherwise.
+     */
+    virtual bool data_exceeds_limitation(
+            CacheChange_t& change,
+            uint32_t size_to_add,
+            uint32_t pending_to_send,
+            RTPSMessageSenderInterface& sender) = 0;
+};
+
+} // namespace rtps
+} // namespace fastdds
+} // namespace eprosima
+
+#endif // FASTDDS_RTPS_MESSAGES__IRTPSMESSAGEGROUPLIMITATION_HPP
+

src/cpp/rtps/messages/RTPSMessageGroup.cpp

@@ -69,21 +69,6 @@ class CacheChangeInlineQoSWriter final : public InlineQosWriter
 
 };
 
-static bool data_exceeds_limitation(
-        uint32_t size_to_add,
-        uint32_t limitation,
-        uint32_t total_sent,
-        uint32_t pending_to_send)
-{
-    return
-        // Limitation has been set and
-        (0 < limitation) &&
-        //   either limitation has already been reached
-        ((limitation <= (total_sent + pending_to_send)) ||
-        //   or adding size_to_add will exceed limitation
-        (size_to_add > (limitation - (total_sent + pending_to_send))));
-}
-
 static bool append_message(
         RTPSParticipantImpl* participant,
         CDRMessage_t* full_msg,
@@ -260,6 +245,7 @@ RTPSMessageGroup::RTPSMessageGroup(
     , max_blocking_time_point_(max_blocking_time_point)
     , send_buffer_(!internal_buffer ? participant->get_send_buffer(max_blocking_time_point) : nullptr)
     , internal_buffer_(internal_buffer)
+    , config_send_buffer_size_(participant->get_min_network_send_buffer_size())
 {
     // Avoid warning when neither SECURITY nor DEBUG is used
     (void)participant;
@@ -406,7 +392,18 @@ void RTPSMessageGroup::send()
             {
                 throw timeout();
             }
-            current_sent_bytes_ += buffers_bytes_;
+
+            current_send_buffer_size_ += buffers_bytes_;
+            if (current_send_buffer_size_ > config_send_buffer_size_)
+            {
+                ++num_of_exceeded_send_buffer_size;
+                current_send_buffer_size_ = current_send_buffer_size_ - config_send_buffer_size_;
+            }
+
+            if (nullptr != limitation_)
+            {
+                limitation_->add_sent_bytes_by_group(buffers_bytes_, *sender_);
+            }
         }
     }
 }
@@ -572,8 +569,7 @@ bool RTPSMessageGroup::add_data(
 
     // Check limitation
     uint32_t data_size = change.serializedPayload.length;
-    if (data_exceeds_limitation(data_size, sent_bytes_limitation_, current_sent_bytes_,
-            buffers_bytes_))
+    if (nullptr != limitation_ && limitation_->data_exceeds_limitation(change, data_size, buffers_bytes_, *sender_))
     {
         flush_and_reset();
         throw limit_exceeded();
@@ -692,8 +688,7 @@ bool RTPSMessageGroup::add_data_frag(
     uint32_t fragment_size = fragment_number < change.getFragmentCount() ? change.getFragmentSize() :
             change.serializedPayload.length - fragment_start;
     // Check limitation
-    if (data_exceeds_limitation(fragment_size, sent_bytes_limitation_, current_sent_bytes_,
-            buffers_bytes_))
+    if (nullptr != limitation_ && limitation_->data_exceeds_limitation(change, fragment_size, buffers_bytes_, *sender_))
     {
         flush_and_reset();
         throw limit_exceeded();

src/cpp/rtps/messages/RTPSMessageGroup.hpp

@@ -32,6 +32,7 @@
 #include <fastdds/rtps/transport/NetworkBuffer.hpp>
 #include <fastdds/utils/collections/ResourceLimitedVector.hpp>
 
+#include <rtps/messages/IRTPSMessageGroupLimitation.hpp>
 #include <rtps/messages/RTPSMessageCreator.hpp>
 
 namespace eprosima {
@@ -232,20 +233,20 @@ class RTPSMessageGroup
         return std::numeric_limits<uint16_t>::max() - data_frag_header_size_ - max_inline_qos_size_ - 3;
     }
 
-    void set_sent_bytes_limitation(
-            uint32_t limit)
+    void set_limitation(
+            IRTPSMessageGroupLimitation* limitation)
     {
-        sent_bytes_limitation_ = limit;
+        limitation_ = limitation;
     }
 
-    void reset_current_bytes_processed()
+    uint32_t current_buffer_bytes() const
     {
-        current_sent_bytes_ = 0;
+        return buffers_bytes_;
     }
 
-    inline uint32_t get_current_bytes_processed() const
+    uint32_t num_exceeded_send_buffer_size() const
     {
-        return current_sent_bytes_ + buffers_bytes_;
+        return num_of_exceeded_send_buffer_size;
     }
 
 private:
@@ -326,17 +327,17 @@ class RTPSMessageGroup
     void add_stats_submsg();
 #endif // FASTDDS_STATISTICS
 
-    RTPSMessageSenderInterface* sender_ = nullptr;
+    RTPSMessageSenderInterface* sender_ {nullptr};
 
-    Endpoint* endpoint_ = nullptr;
+    Endpoint* endpoint_ {nullptr};
 
-    CDRMessage_t* header_msg_ = nullptr;
+    CDRMessage_t* header_msg_ {nullptr};
 
-    CDRMessage_t* submessage_msg_ = nullptr;
+    CDRMessage_t* submessage_msg_ {nullptr};
 
     GuidPrefix_t current_dst_;
 
-    RTPSParticipantImpl* participant_ = nullptr;
+    RTPSParticipantImpl* participant_ {nullptr};
 
      #if HAVE_SECURITY
 
@@ -348,11 +349,7 @@ class RTPSMessageGroup
 
     std::unique_ptr<RTPSMessageGroup_t> send_buffer_;
 
-    bool internal_buffer_ = false;
-
-    uint32_t sent_bytes_limitation_ = 0;
-
-    uint32_t current_sent_bytes_ = 0;
+    bool internal_buffer_ {false};
 
     // Next buffer that will be sent
     eprosima::fastdds::rtps::NetworkBuffer pending_buffer_;
@@ -363,6 +360,12 @@ class RTPSMessageGroup
     // Vector of payloads of which the RTPSMessageGroup is the owner
     ResourceLimitedVector<eprosima::fastdds::rtps::SerializedPayload_t>* payloads_to_send_ = nullptr;
 
+    const uint32_t config_send_buffer_size_ {0};
+
+    uint32_t current_send_buffer_size_ {0};
+
+    uint32_t num_of_exceeded_send_buffer_size {0};
+
     // Bytes to send in the next list of buffers
     uint32_t buffers_bytes_ = 0;
 
@@ -371,6 +374,8 @@ class RTPSMessageGroup
 
     // Fixed padding to be used whenever needed
     const octet padding_[3] = {0, 0, 0};
+
+    IRTPSMessageGroupLimitation* limitation_ {nullptr};
 };
 
 } // namespace rtps

src/cpp/rtps/participant/RTPSParticipantImpl.hpp

@@ -48,7 +48,7 @@
 #include <fastdds/rtps/history/IPayloadPool.hpp>
 #include <fastdds/rtps/transport/SenderResource.hpp>
 
-#include "../flowcontrol/FlowControllerFactory.hpp"
+#include <fastdds/types.hpp>
 #include <fastdds/utils/TypePropagation.hpp>
 #include <rtps/builtin/data/ReaderProxyData.hpp>
 #include <rtps/builtin/data/WriterProxyData.hpp>
@@ -746,7 +746,7 @@ class RTPSParticipantImpl
     /*
      * Flow controller factory.
      */
-    FlowControllerFactory flow_controller_factory_;
+    FlowControllerFactoryType flow_controller_factory_;
 
 #if HAVE_SECURITY
     security::ParticipantSecurityAttributes security_attributes_;