[25464] Implement payload release in move assignment

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

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;
         }
 

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<void>(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<void>(success);
+        assert(success);
+        payload_owner = nullptr;
     }
     this->empty();
 }

src/cpp/rtps/messages/MessageReceiver.cpp

@@ -47,6 +47,38 @@ namespace eprosima {
 namespace fastdds {
 namespace rtps {
 
+struct NoOpPayloadPool : public IPayloadPool
+{
+    bool get_payload(
+        uint32_t size,
+        SerializedPayload_t& payload) override
+    {
+        static_cast<void>(payload);
+        static_cast<void>(size);
+        return false;
+    }
+
+    bool get_payload(
+            const SerializedPayload_t& data,
+            SerializedPayload_t& payload) override
+    {
+        static_cast<void>(data);
+        static_cast<void>(payload);
+        return false;
+    }
+
+    bool release_payload(
+            SerializedPayload_t& payload) override
+    {
+        if (payload.payload_owner == this)
+        {
+            payload.data = nullptr;
+            return true;
+        }
+        return false;
+    }
+};
+
 MessageReceiver::MessageReceiver(
         RTPSParticipantImpl* participant,
         uint32_t rec_buffer_size)
@@ -796,6 +828,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 +879,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 +906,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 +932,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 +985,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 +1050,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 +1074,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 +1099,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");
 

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
+    $<$<AND:$<NOT:$<BOOL:${WIN32}>>,$<STREQUAL:"${CMAKE_BUILD_TYPE}","Debug">>:__DEBUG>
+    $<$<BOOL:${INTERNAL_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
     $<$<AND:$<NOT:$<BOOL:${WIN32}>>,$<STREQUAL:"${CMAKE_BUILD_TYPE}","Debug">>:__DEBUG>

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 <cstddef>
+#include <cstdlib>
+#include <utility>
+
+#include <gtest/gtest.h>
+
+#include <fastdds/rtps/common/SerializedPayload.hpp>
+
+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<octet*>(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();
+}