diff --git a/include/fastdds/rtps/common/SerializedPayload.hpp b/include/fastdds/rtps/common/SerializedPayload.hpp index c856f23a3bd..663fd887a1f 100644 --- a/include/fastdds/rtps/common/SerializedPayload.hpp +++ b/include/fastdds/rtps/common/SerializedPayload.hpp @@ -103,10 +103,27 @@ struct FASTDDS_EXPORTED_API SerializedPayload_t const SerializedPayload_t& other) = delete; //!Move constructor + //!Directly moves the fields from \c other instead of calling the move assignment operator, + //!since the latter assumes the destination is already constructed (it may release the + //!currently owned payload). Using the move assignment operator here would read + //!uninitialized members of the newly constructed object. SerializedPayload_t( SerializedPayload_t&& other) noexcept + : encapsulation(other.encapsulation) + , length(other.length) + , data(other.data) + , max_size(other.max_size) + , pos(other.pos) + , payload_owner(other.payload_owner) + , is_serialized_key(other.is_serialized_key) { - *this = std::move(other); + other.encapsulation = CDR_BE; + other.length = 0; + other.data = nullptr; + other.max_size = 0; + other.pos = 0; + other.payload_owner = nullptr; + other.is_serialized_key = false; } //!Move operator diff --git a/src/cpp/rtps/DataSharing/ReaderPool.hpp b/src/cpp/rtps/DataSharing/ReaderPool.hpp index d676c98f190..08f24a48863 100644 --- a/src/cpp/rtps/DataSharing/ReaderPool.hpp +++ b/src/cpp/rtps/DataSharing/ReaderPool.hpp @@ -255,6 +255,7 @@ class ReaderPool : public DataSharingPayloadPool if (check == c_SequenceNumber_Unknown || check != cache_change.sequenceNumber) { // data override while processing + cache_change.serializedPayload.data = nullptr; return false; } diff --git a/src/cpp/rtps/common/SerializedPayload.cpp b/src/cpp/rtps/common/SerializedPayload.cpp index 98ea4738373..0675251551c 100644 --- a/src/cpp/rtps/common/SerializedPayload.cpp +++ b/src/cpp/rtps/common/SerializedPayload.cpp @@ -30,6 +30,18 @@ SerializedPayload_t& SerializedPayload_t::operator = ( return *this; } + if (payload_owner != nullptr) + { + bool success = payload_owner->release_payload(*this); + static_cast(success); + assert(success); + payload_owner = nullptr; + } + else if (data != nullptr) + { + free(data); + } + encapsulation = other.encapsulation; length = other.length; data = other.data; @@ -53,7 +65,10 @@ SerializedPayload_t::~SerializedPayload_t() { if (payload_owner != nullptr) { - payload_owner->release_payload(*this); + bool success = payload_owner->release_payload(*this); + static_cast(success); + assert(success); + payload_owner = nullptr; } this->empty(); } diff --git a/src/cpp/rtps/history/NoOpPayloadPool.hpp b/src/cpp/rtps/history/NoOpPayloadPool.hpp new file mode 100644 index 00000000000..8344541b10b --- /dev/null +++ b/src/cpp/rtps/history/NoOpPayloadPool.hpp @@ -0,0 +1,64 @@ +// 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 NoOpPayloadPool.hpp + */ + +#ifndef RTPS_HISTORY_NOOPPAYLOADPOOL_HPP +#define RTPS_HISTORY_NOOPPAYLOADPOOL_HPP + +#include + +namespace eprosima { +namespace fastdds { +namespace rtps { + +struct NoOpPayloadPool : public IPayloadPool +{ + bool get_payload( + uint32_t size, + SerializedPayload_t& payload) override + { + static_cast(payload); + static_cast(size); + return false; + } + + bool get_payload( + const SerializedPayload_t& data, + SerializedPayload_t& payload) override + { + static_cast(data); + static_cast(payload); + return false; + } + + bool release_payload( + SerializedPayload_t& payload) override + { + if (payload.payload_owner == this) + { + payload.data = nullptr; + return true; + } + return false; + } +}; + +} // namespace rtps +} // namespace fastdds +} // namespace eprosima + +#endif // RTPS_HISTORY_NOOPPAYLOADPOOL_HPP diff --git a/src/cpp/rtps/messages/MessageReceiver.cpp b/src/cpp/rtps/messages/MessageReceiver.cpp index f1be51957c5..312bb0fe0b3 100644 --- a/src/cpp/rtps/messages/MessageReceiver.cpp +++ b/src/cpp/rtps/messages/MessageReceiver.cpp @@ -29,6 +29,7 @@ #include #include +#include #include #include #include @@ -796,6 +797,7 @@ bool MessageReceiver::proc_Submsg_Data( //FOUND THE READER. //We ask the reader for a cachechange to store the information. + NoOpPayloadPool no_op_pool; CacheChange_t ch; ch.kind = ALIVE; ch.writerGUID.guidPrefix = source_guid_prefix_; @@ -846,6 +848,7 @@ bool MessageReceiver::proc_Submsg_Data( ch.inline_qos.length = inlineQosSize; ch.inline_qos.encapsulation = endiannessFlag ? PL_CDR_LE : PL_CDR_BE; ch.inline_qos.pos = 0; + ch.inline_qos.payload_owner = &no_op_pool; } if (dataFlag || keyFlag) @@ -872,6 +875,7 @@ bool MessageReceiver::proc_Submsg_Data( ch.serializedPayload.length = payload_size; ch.serializedPayload.max_size = payload_size; ch.serializedPayload.is_serialized_key = keyFlag; + ch.serializedPayload.payload_owner = &no_op_pool; msg->pos = next_pos; } else @@ -897,16 +901,6 @@ bool MessageReceiver::proc_Submsg_Data( //Look for the correct reader to add the change process_data_message_function_(readerID, ch, was_decoded); - IPayloadPool* payload_pool = ch.serializedPayload.payload_owner; - if (payload_pool) - { - payload_pool->release_payload(ch.serializedPayload); - } - - //TODO(Ricardo) If an exception is thrown (ex, by fastcdr), these lines are not executed -> segmentation fault - ch.serializedPayload.data = nullptr; - ch.inline_qos.data = nullptr; - EPROSIMA_LOG_INFO(RTPS_MSG_IN, IDSTRING "Sub Message DATA processed"); return true; } @@ -960,6 +954,7 @@ bool MessageReceiver::proc_Submsg_DataFrag( //FOUND THE READER. //We ask the reader for a cachechange to store the information. + NoOpPayloadPool no_op_pool; CacheChange_t ch; ch.kind = ALIVE; ch.writerGUID.guidPrefix = source_guid_prefix_; @@ -1024,6 +1019,7 @@ bool MessageReceiver::proc_Submsg_DataFrag( ch.inline_qos.length = inlineQosSize; ch.inline_qos.encapsulation = endiannessFlag ? PL_CDR_LE : PL_CDR_BE; ch.inline_qos.pos = 0; + ch.inline_qos.payload_owner = &no_op_pool; } uint32_t payload_size; @@ -1047,6 +1043,7 @@ bool MessageReceiver::proc_Submsg_DataFrag( ch.serializedPayload.length = payload_size; ch.serializedPayload.max_size = payload_size; ch.serializedPayload.is_serialized_key = keyFlag; + ch.serializedPayload.payload_owner = &no_op_pool; ch.setFragmentSize(fragmentSize); msg->pos = next_pos; @@ -1071,8 +1068,6 @@ bool MessageReceiver::proc_Submsg_DataFrag( << associated_readers_.size()); process_data_fragment_message_function_(readerID, ch, sampleSize, fragmentStartingNum, fragmentsInSubmessage, was_decoded); - ch.serializedPayload.data = nullptr; - ch.inline_qos.data = nullptr; EPROSIMA_LOG_INFO(RTPS_MSG_IN, IDSTRING "Sub Message DATA_FRAG processed"); diff --git a/src/cpp/rtps/reader/StatefulReader.cpp b/src/cpp/rtps/reader/StatefulReader.cpp index 152719cc09f..fc0d06c5a54 100644 --- a/src/cpp/rtps/reader/StatefulReader.cpp +++ b/src/cpp/rtps/reader/StatefulReader.cpp @@ -37,6 +37,7 @@ #include #include #include +#include #include #include #include @@ -663,7 +664,8 @@ bool StatefulReader::process_data_msg( } else if (payload_pool_->get_payload(change->serializedPayload, change_to_add->serializedPayload)) { - if (change->serializedPayload.payload_owner == nullptr) + if ((nullptr == change->serializedPayload.payload_owner) || + (nullptr != dynamic_cast(change->serializedPayload.payload_owner))) { payload_pool_->get_payload(change_to_add->serializedPayload, change->serializedPayload); } diff --git a/src/cpp/rtps/reader/StatelessReader.cpp b/src/cpp/rtps/reader/StatelessReader.cpp index e9ab68cc39c..eba84cc1d01 100644 --- a/src/cpp/rtps/reader/StatelessReader.cpp +++ b/src/cpp/rtps/reader/StatelessReader.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -676,7 +677,8 @@ bool StatelessReader::process_data_msg( } else if (payload_pool_->get_payload(change->serializedPayload, change_to_add->serializedPayload)) { - if (change->serializedPayload.payload_owner == nullptr) + if ((nullptr == change->serializedPayload.payload_owner) || + (nullptr != dynamic_cast(change->serializedPayload.payload_owner))) { payload_pool_->get_payload(change_to_add->serializedPayload, change->serializedPayload); } diff --git a/test/unittest/rtps/common/CMakeLists.txt b/test/unittest/rtps/common/CMakeLists.txt index f7376fcf2ff..b0707a058d3 100644 --- a/test/unittest/rtps/common/CMakeLists.txt +++ b/test/unittest/rtps/common/CMakeLists.txt @@ -45,6 +45,8 @@ if(ANDROID) endif() +set(SERIALIZEDPAYLOADTESTS_SOURCE SerializedPayloadTests.cpp + ${PROJECT_SOURCE_DIR}/src/cpp/rtps/common/SerializedPayload.cpp) set(SEQUENCENUMBERTESTS_SOURCE SequenceNumberTests.cpp) set(PORTPARAMETERSTESTS_SOURCE PortParametersTests.cpp ${PROJECT_SOURCE_DIR}/src/cpp/fastdds/core/Time_t.cpp @@ -100,6 +102,16 @@ target_link_libraries(GuidUtilsTests ) gtest_discover_tests(GuidUtilsTests) +add_executable(SerializedPayloadTests ${SERIALIZEDPAYLOADTESTS_SOURCE}) +target_compile_definitions(SerializedPayloadTests PRIVATE + $<$>,$>:__DEBUG> + $<$:__INTERNALDEBUG> # Internal debug activated. + ) +target_include_directories(SerializedPayloadTests PRIVATE + ${PROJECT_SOURCE_DIR}/include ${PROJECT_BINARY_DIR}/include) +target_link_libraries(SerializedPayloadTests GTest::gtest) +gtest_discover_tests(SerializedPayloadTests) + add_executable(SequenceNumberTests ${SEQUENCENUMBERTESTS_SOURCE}) target_compile_definitions(SequenceNumberTests PRIVATE $<$>,$>:__DEBUG> diff --git a/test/unittest/rtps/common/SerializedPayloadTests.cpp b/test/unittest/rtps/common/SerializedPayloadTests.cpp new file mode 100644 index 00000000000..ff839373099 --- /dev/null +++ b/test/unittest/rtps/common/SerializedPayloadTests.cpp @@ -0,0 +1,131 @@ +// Copyright 2026 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. + +#include +#include +#include + +#include + +#include + +using namespace eprosima::fastdds::rtps; + +class TestingPayloadPool : public IPayloadPool +{ +public: + + ~TestingPayloadPool() override = default; + + bool get_payload( + uint32_t size, + SerializedPayload_t& payload) override + { + payload.data = static_cast(calloc(size, sizeof(octet))); + if (payload.data == nullptr) + { + return false; + } + + payload.payload_owner = this; + payload.max_size = size; + payload.length = size; + return true; + } + + bool get_payload( + const SerializedPayload_t& data, + SerializedPayload_t& payload) override + { + if (!get_payload(data.length, payload)) + { + return false; + } + + payload.copy(&data); + return true; + } + + bool release_payload( + SerializedPayload_t& payload) override + { + ++release_payload_calls; + EXPECT_EQ(this, payload.payload_owner); + + free(payload.data); + payload.data = nullptr; + payload.payload_owner = nullptr; + payload.length = 0; + payload.max_size = 0; + return true; + } + + size_t release_payload_calls = 0; +}; + +/*! + * @fn TEST(SerializedPayload, MoveAssignmentOperatorReleasesCurrentPayload) + * @brief This test checks that move assignment operator releases an already owned destination payload. + */ +TEST(SerializedPayload, MoveAssignmentOperatorReleasesCurrentPayload) +{ + TestingPayloadPool payload_pool; + SerializedPayload_t destination; + ASSERT_TRUE(payload_pool.get_payload(4, destination)); + + SerializedPayload_t source(8); + source.length = 8; + + destination = std::move(source); + + EXPECT_EQ(1u, payload_pool.release_payload_calls); + EXPECT_EQ(8u, destination.length); + EXPECT_EQ(8u, destination.max_size); + EXPECT_NE(nullptr, destination.data); + EXPECT_EQ(nullptr, destination.payload_owner); + EXPECT_EQ(0u, source.length); + EXPECT_EQ(0u, source.max_size); + EXPECT_EQ(nullptr, source.data); +} + +/*! + * @fn TEST(SerializedPayload, MoveAssignmentOperatorReleasesCurrentUnownedPayload) + * @brief This test checks that move assignment operator releases an already allocated destination payload. + */ +TEST(SerializedPayload, MoveAssignmentOperatorReleasesCurrentUnownedPayload) +{ + SerializedPayload_t destination(4); + ASSERT_NE(nullptr, destination.data); + + SerializedPayload_t source(8); + source.length = 8; + + destination = std::move(source); + + EXPECT_EQ(8u, destination.length); + EXPECT_EQ(8u, destination.max_size); + EXPECT_NE(nullptr, destination.data); + EXPECT_EQ(nullptr, destination.payload_owner); + EXPECT_EQ(0u, source.length); + EXPECT_EQ(0u, source.max_size); + EXPECT_EQ(nullptr, source.data); +} + +int main( + int argc, + char** argv) +{ + testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}