GAP allocation improvements (#1031)

* GAP allocation improvements <master>

* Refs #6169. Added RTPSGapBuilder to avoid vector of pairs.

* Refs #6169. Using RTPSGapBuilder when sending gaps to single reader.

* Refs #6169. Using RTPSGapBuilder to send gaps related to holes in history.

* Refs #6169. Fixing rebase issues.

* Refs #6169. Removing tabs.

* Refs #6169. Destination specific versions of some methods of RTPSMessageGroup

* Refs #6169. Destination specific version of RTPSMessageGroup::add_gap.

* Refs #6169. Destination specific option on RTPSGapBuilder.

* Refs #6169. Using destination specific gaps on StatefulWriter.

* Refs #6169. Removed StatefulWriterOrganizer.

* Refs #7070. Minor suggestions from code review

Co-Authored-By: Ricardo González <correoricky@gmail.com>

* Refs #7070. Avoid sending gaps for sequence numbers under min in history.

* Refs #7070. Sending heartbeat before gaps.

* Refs #7070. No need of forcing piggyback heartbeats.

* Refs #7070. Fixing warning.

* Refs #7070. Only sending gaps to reliable readers.

* Refs #7070. Not building gaps with empty history.

Co-authored-by: Ricardo González <correoricky@gmail.com>

* Refs #7071. GAP generation algorithm improved.

* Refs #7071. Additional check to avoid unnecessary lookups.

* Apply suggestions from code review

Co-authored-by: Ricardo González <correoricky@gmail.com>
Co-authored-by: Miguel Company <miguelcompany@eprosima.com>

include/fastdds/rtps/messages/RTPSMessageGroup.h

@@ -110,11 +110,34 @@ class Endpoint;
                 bool liveliness_flag);
 
         /**
-         * Adds a GAP message to the group.
+         * Adds one or more GAP messages to the group.
          * @param changes_seq_numbers Set of missed sequence numbers.
+         * @return True when messages were added to the group.
+         */
+        bool add_gap(
+                std::set<SequenceNumber_t>& changes_seq_numbers);
+
+        /**
+         * Adds one GAP message to the group.
+         * @param gap_initial_sequence Start of consecutive sequence numbers.
+         * @param gap_bitmap Bitmap of non-consecutive sequence numbers.
+         * @return True when message was added to the group.
+         */
+        bool add_gap(
+                const SequenceNumber_t& gap_initial_sequence,
+                const SequenceNumberSet_t& gap_bitmap);
+
+        /**
+         * Adds one GAP message to the group.
+         * @param gap_initial_sequence Start of consecutive sequence numbers.
+         * @param gap_bitmap Bitmap of non-consecutive sequence numbers.
+         * @param reader_guid GUID of the destination reader.
          * @return True when message was added to the group.
          */
-        bool add_gap(std::set<SequenceNumber_t>& changes_seq_numbers);
+        bool add_gap(
+                const SequenceNumber_t& gap_initial_sequence,
+                const SequenceNumberSet_t& gap_bitmap,
+                const GUID_t& reader_guid);
 
         /**
          * Adds a ACKNACK message to the group.
@@ -159,14 +182,33 @@ class Endpoint;
 
         void send();
 
-        void check_and_maybe_flush();
+        void check_and_maybe_flush()
+        {
+            check_and_maybe_flush(sender_.destination_guid_prefix());
+        }
+
+        void check_and_maybe_flush(
+                const GuidPrefix_t& destination_guid_prefix);
 
-        bool insert_submessage();
+        bool insert_submessage()
+        {
+            return insert_submessage(sender_.destination_guid_prefix());
+        }
+
+        bool insert_submessage(
+                const GuidPrefix_t& destination_guid_prefix);
 
-        bool add_info_dst_in_buffer(CDRMessage_t* buffer);
+        bool add_info_dst_in_buffer(
+                CDRMessage_t* buffer,
+                const GuidPrefix_t& destination_guid_prefix);
 
         bool add_info_ts_in_buffer(const Time_t& timestamp);
 
+        bool create_gap_submessage(
+                const SequenceNumber_t& gap_initial_sequence,
+                const SequenceNumberSet_t& gap_bitmap,
+                const EntityId_t& reader_id);
+
         const RTPSMessageSenderInterface& sender_;
 
         Endpoint* endpoint_;

include/fastdds/rtps/writer/StatefulWriter.h

@@ -310,8 +310,7 @@ class StatefulWriter : public RTPSWriter
     void send_heartbeat_piggyback_nts_(
             ReaderProxy* reader,
             RTPSMessageGroup& message_group,
-            uint32_t& last_bytes_processed,
-            bool force = false);
+            uint32_t& last_bytes_processed);
 
     void send_heartbeat_nts_(
             size_t number_of_readers,
@@ -335,6 +334,8 @@ class StatefulWriter : public RTPSWriter
     std::chrono::duration<double, std::ratio<1, 1000000> > keep_duration_us_;
     //! Last acknowledged cache change (only used if using disable positive ACKs QoS)
     SequenceNumber_t last_sequence_number_;
+    //! Biggest sequence number removed from history
+    SequenceNumber_t biggest_removed_sequence_number_;
 
     const uint32_t sendBufferSize_;
 

src/cpp/CMakeLists.txt

@@ -62,6 +62,7 @@ set(${PROJECT_NAME}_source_files
     rtps/reader/RTPSReader.cpp
     rtps/messages/RTPSMessageCreator.cpp
     rtps/messages/RTPSMessageGroup.cpp
+    rtps/messages/RTPSGapBuilder.cpp
     rtps/messages/MessageReceiver.cpp
     rtps/messages/submessages/AckNackMsg.hpp
     rtps/messages/submessages/DataMsg.hpp

src/cpp/rtps/messages/RTPSGapBuilder.cpp

@@ -0,0 +1,86 @@
+// Copyright 2020 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 RTPSGapBuilder.cpp
+ *
+ */
+
+#include "RTPSGapBuilder.hpp"
+
+namespace eprosima {
+namespace fastrtps {
+namespace rtps {
+
+RTPSGapBuilder::~RTPSGapBuilder()
+{
+    flush();
+}
+
+bool RTPSGapBuilder::add(
+        const SequenceNumber_t& gap_sequence)
+{
+    // Check if it is the first gap being added
+    if (!is_gap_pending_)
+    {
+        is_gap_pending_ = true;
+        initial_sequence_ = gap_sequence;
+        gap_bitmap_.base(gap_sequence + 1);
+        return true;
+    }
+
+    // Check for contiguous from initial_sequence_
+    SequenceNumber_t base = gap_bitmap_.base();
+    if (gap_sequence == base)
+    {
+        gap_bitmap_.base(gap_sequence + 1);
+        return true;
+    }
+
+    // Check if past last in bitmap
+    if (gap_bitmap_.add(gap_sequence))
+    {
+        return true;
+    }
+
+    // Did not fit inside bitmap. Difference between gap_sequence and base is greater than 255.
+    // Send GAP with current info and prepare info for next GAP.
+    bool ret_val = flush();
+    is_gap_pending_ = true;
+    initial_sequence_ = gap_sequence;
+    gap_bitmap_.base(gap_sequence + 1);
+
+    return ret_val;
+}
+
+bool RTPSGapBuilder::flush()
+{
+    if (is_gap_pending_)
+    {
+        bool ok = with_specific_destination_ ?
+            group_.add_gap(initial_sequence_, gap_bitmap_, reader_guid_) :
+            group_.add_gap(initial_sequence_, gap_bitmap_);
+        if (!ok)
+        {
+            return false;
+        }
+    }
+
+    is_gap_pending_ = false;
+    return true;
+}
+
+} // namespace rtps
+} // namespace fastrtps
+} // namespace eprosima

src/cpp/rtps/messages/RTPSGapBuilder.hpp

@@ -0,0 +1,107 @@
+// Copyright 2020 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 RTPSGapBuilder.hpp
+ *
+ */
+
+#ifndef RTPSGAPBUILDER_HPP
+#define RTPSGAPBUILDER_HPP
+#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
+
+#include <fastrtps/rtps/messages/RTPSMessageGroup.h>
+
+namespace eprosima {
+namespace fastrtps {
+namespace rtps {
+
+/**
+ * A helper class to add GAP messages to a @ref RTPSMessageGroup.
+ * @ingroup WRITER_MODULE
+ */
+class RTPSGapBuilder
+{
+public:
+
+    /**
+     * RTPSGapBuilder constructor.
+     *
+     * @param group Reference to the @ref RTPSMessageGroup that will be used to send GAP messages.
+     */
+    explicit RTPSGapBuilder(
+            RTPSMessageGroup& group)
+        : group_(group)
+    {
+    }
+
+    /**
+     * RTPSGapBuilder constructor.
+     *
+     * @param group Reference to the @ref RTPSMessageGroup that will be used to send GAP messages.
+     * @param reader_guid Specific destination reader guid
+     */
+    explicit RTPSGapBuilder(
+            RTPSMessageGroup& group,
+            const GUID_t& reader_guid)
+        : with_specific_destination_(true)
+        , group_(group)
+        , reader_guid_(reader_guid)
+    {
+    }
+
+    ~RTPSGapBuilder();
+
+    /**
+     * Adds a sequence number to the GAP list.
+     *
+     * @remark Sequence numbers should be added in strict increasing order.
+     * 
+     * @param gap_sequence Sequence number to be added to the GAP list.
+     * @return false if a GAP message couldn't be added to the message group,
+     *         true if no GAP message was needed or it was successfully added.
+     *
+     * @throws RTPSMessageGroup::timeout if a network operation was necessary and
+     *         it blocked for more than the maximum time allowed.
+     */
+    bool add(
+            const SequenceNumber_t& gap_sequence);
+
+    /**
+     * Adds a GAP message to the message group if necessary.
+     *
+     * @return false if a GAP message couldn't be added to the message group,
+     *         true if no GAP message was needed or it was successfully added.
+     *
+     * @throws RTPSMessageGroup::timeout if a network operation was necessary and
+     *         it blocked for more than the maximum time allowed.
+     */
+    bool flush();
+
+private:
+
+    bool is_gap_pending_ = false;             ///< Whether a GAP message is pending to be added.
+    bool with_specific_destination_ = false;  ///< Whether a specific reader_guid is used.
+    RTPSMessageGroup& group_;                 ///< Reference to the message group used to output messages.
+    SequenceNumber_t initial_sequence_;       ///< Contiguous range initial sequence.
+    SequenceNumberSet_t gap_bitmap_;          ///< Bitmap with non-contiguous sequences.
+    GUID_t reader_guid_;                      ///< Specific destination reader guid.
+};
+
+} /* namespace rtps */
+} /* namespace fastrtps */
+} /* namespace eprosima */
+
+#endif
+#endif /* RTPSGAPBUILDER_HPP */