diff --git a/src/cpp/rtps/DataSharing/DataSharingListener.cpp b/src/cpp/rtps/DataSharing/DataSharingListener.cpp index f6b174df41a..f76aa96b5e2 100644 --- a/src/cpp/rtps/DataSharing/DataSharingListener.cpp +++ b/src/cpp/rtps/DataSharing/DataSharingListener.cpp @@ -157,16 +157,16 @@ void DataSharingListener::process_new_data () pool->get_next_unread_payload(ch, last_sequence, last_payload); has_new_payload = ch.sequenceNumber != c_SequenceNumber_Unknown; - if (has_new_payload) + if (has_new_payload && ch.sequenceNumber > SequenceNumber_t(0, 0)) { - if (last_sequence != c_SequenceNumber_Unknown && ch.sequenceNumber != last_sequence + 1) + if (last_sequence != c_SequenceNumber_Unknown && ch.sequenceNumber > last_sequence + 1) { EPROSIMA_LOG_WARNING(RTPS_READER, "GAP (" << last_sequence + 1 << " - " << ch.sequenceNumber - 1 << ")" << " detected on datasharing writer " << pool->writer()); reader_->processGapMsg(pool->writer(), last_sequence + 1, SequenceNumberSet_t(ch.sequenceNumber)); } - if (last_sequence == c_SequenceNumber_Unknown && ch.sequenceNumber != SequenceNumber_t(0, 1)) + if (last_sequence == c_SequenceNumber_Unknown && ch.sequenceNumber > SequenceNumber_t(0, 1)) { EPROSIMA_LOG_INFO(RTPS_READER, "First change with SN " << ch.sequenceNumber << " detected on datasharing writer " << diff --git a/src/cpp/rtps/DataSharing/ReaderPool.hpp b/src/cpp/rtps/DataSharing/ReaderPool.hpp index c5413a738c6..a0aa47acedb 100644 --- a/src/cpp/rtps/DataSharing/ReaderPool.hpp +++ b/src/cpp/rtps/DataSharing/ReaderPool.hpp @@ -192,6 +192,12 @@ class ReaderPool : public DataSharingPayloadPool continue; } + if (last_sn_ != c_SequenceNumber_Unknown && last_sn_ >= cache_change.sequenceNumber) + { + // Sequence number went backwards, it was most probably overriden. + continue; + } + if (!ensure_reading_reference_is_in_bounds()) { // We may have been taken over and read a payload that is too far forward. Discard and continue diff --git a/src/cpp/rtps/reader/StatefulReader.cpp b/src/cpp/rtps/reader/StatefulReader.cpp index c1232442e48..04216834820 100644 --- a/src/cpp/rtps/reader/StatefulReader.cpp +++ b/src/cpp/rtps/reader/StatefulReader.cpp @@ -851,7 +851,7 @@ bool StatefulReader::processGapMsg( WriterProxy* pWP = nullptr; std::unique_lock lock(mp_mutex); - if (!is_alive_) + if (!is_alive_ || gapStart < SequenceNumber_t(0, 1) || gapList.base() <= gapStart) { return false; } @@ -860,9 +860,9 @@ bool StatefulReader::processGapMsg( { // TODO (Miguel C): Refactor this inside WriterProxy SequenceNumber_t auxSN; - SequenceNumber_t finalSN = gapList.base() - 1; + SequenceNumber_t finalSN = gapList.base(); History::const_iterator history_iterator = mp_history->changesBegin(); - for (auxSN = gapStart; auxSN <= finalSN; auxSN++) + for (auxSN = gapStart; auxSN < finalSN; auxSN++) { if (pWP->irrelevant_change_set(auxSN)) { diff --git a/test/blackbox/common/RTPSBlackboxTestsBasic.cpp b/test/blackbox/common/RTPSBlackboxTestsBasic.cpp index f5878fab0b1..d1348d8a98c 100644 --- a/test/blackbox/common/RTPSBlackboxTestsBasic.cpp +++ b/test/blackbox/common/RTPSBlackboxTestsBasic.cpp @@ -866,6 +866,35 @@ TEST(RTPS, MultithreadedWriterCreation) RTPSDomain::stopAll(); } +/* Regression Test for improving gaps processing + * https://github.com/eProsima/Fast-DDS/pull/3343 + */ +TEST(RTPS, RTPSCorrectGAPProcessing) +{ + RTPSWithRegistrationReader reader(TEST_TOPIC_NAME); + RTPSWithRegistrationWriter writer(TEST_TOPIC_NAME); + + reader.durability(eprosima::fastrtps::rtps::DurabilityKind_t::TRANSIENT_LOCAL). + reliability(eprosima::fastrtps::rtps::ReliabilityKind_t::RELIABLE).init(); + + ASSERT_TRUE(reader.isInitialized()); + + writer.durability(eprosima::fastrtps::rtps::DurabilityKind_t::TRANSIENT_LOCAL). + reliability(eprosima::fastrtps::rtps::ReliabilityKind_t::RELIABLE).init(); + + ASSERT_TRUE(writer.isInitialized()); + + reader.wait_discovery(); + writer.wait_discovery(); + + SequenceNumberSet_t seq_set(SequenceNumber_t(0, 0)); + + //! GAP Message check + RTPSReader& native_reader = reader.get_native_reader(); + ASSERT_NO_FATAL_FAILURE(native_reader.processGapMsg(writer.guid(), {0, 0}, seq_set)); +} + + class CustomReaderDataFilter : public eprosima::fastdds::rtps::IReaderDataFilter { public: diff --git a/test/blackbox/common/RTPSWithRegistrationReader.hpp b/test/blackbox/common/RTPSWithRegistrationReader.hpp index 4ddb1bc2956..414ee069555 100644 --- a/test/blackbox/common/RTPSWithRegistrationReader.hpp +++ b/test/blackbox/common/RTPSWithRegistrationReader.hpp @@ -527,6 +527,11 @@ class RTPSWithRegistrationReader return reader_->getGuid(); } + eprosima::fastrtps::rtps::RTPSReader& get_native_reader() const + { + return *reader_; + } + private: void receive_one(