Skip to content

Commit 20606d1

Browse files
tempatejuanlofer-eprosima
tempate
and
juanlofer-eprosima
authored
Share a custom PayloadPool between DDS and RTPS participants (#70)
* Use the router's PayloadPool to avoid copies Signed-off-by: tempate <danieldiaz@eprosima.com> * Update TypeSupport to use the PayloadPool Signed-off-by: tempate <danieldiaz@eprosima.com> * Mediator PoC: no leaks and copies Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com> * Fix memory leaks: XML participants without ignore_local_endpoints Signed-off-by: tempate <danieldiaz@eprosima.com> * Replace the PoC with the actual implementation Signed-off-by: tempate <danieldiaz@eprosima.com> * Destroy saved payloads after getting them Signed-off-by: tempate <danieldiaz@eprosima.com> * Destroy the discarded payloads Signed-off-by: tempate <danieldiaz@eprosima.com> * Uncrustify Signed-off-by: tempate <danieldiaz@eprosima.com> * Apply suggestions Signed-off-by: tempate <danieldiaz@eprosima.com> * Complete TODOs Signed-off-by: tempate <danieldiaz@eprosima.com> * Apply suggestions Signed-off-by: tempate <danieldiaz@eprosima.com> * Apply more suggestions Signed-off-by: tempate <danieldiaz@eprosima.com> --------- Signed-off-by: tempate <danieldiaz@eprosima.com> Signed-off-by: Juan Lopez Fernandez <juanlopez@eprosima.com> Co-authored-by: Juan Lopez Fernandez <juanlopez@eprosima.com>
1 parent 5579d49 commit 20606d1

8 files changed

Lines changed: 326 additions & 23 deletions

File tree

‎ddspipe_core/include/ddspipe_core/efficiency/payload/PayloadPoolMediator.hpp‎

Lines changed: 183 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,183 @@
1+
// Copyright 2023 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
#pragma once
16+
17+
#include <atomic>
18+
#include <map>
19+
#include <mutex>
20+
21+
#include <cpp_utils/ReturnCode.hpp>
22+
23+
#include <fastdds/dds/publisher/DataWriter.hpp>
24+
#include <fastdds/rtps/common/InstanceHandle.h>
25+
#include <fastdds/rtps/common/SerializedPayload.h>
26+
#include <fastdds/rtps/history/IPayloadPool.h>
27+
28+
#include <ddspipe_core/efficiency/payload/PayloadPool.hpp>
29+
#include <ddspipe_core/types/data/RtpsPayloadData.hpp>
30+
31+
namespace eprosima {
32+
namespace ddspipe {
33+
namespace core {
34+
35+
/**
36+
* DataWriters always allocate memory for a payload when writing it to the \c PayloadPool. In the DdsPipe, however, the
37+
* \c payload is already in the \c PayloadPool (saved by the Reader), so we can reuse it to avoid an extra copy.
38+
*
39+
* This class works as a mediator between the \c FastPayloadPool and the DataWriter's write functions, to avoid making
40+
* the extra copy.
41+
*/
42+
class PayloadPoolMediator : public fastrtps::rtps::IPayloadPool
43+
{
44+
public:
45+
46+
/**
47+
* @brief construct a \c PayloadPoolMediator to mediate for a \c PayloadPool.
48+
*
49+
* @param payload_pool the \c PayloadPool to mediate for.
50+
*/
51+
DDSPIPE_CORE_DllAPI
52+
PayloadPoolMediator(
53+
const std::shared_ptr<PayloadPool>& payload_pool);
54+
55+
/**
56+
* @brief write \c data with a \c writer but avoiding copying the \c data from and to the same \c PayloadPool.
57+
*
58+
* Save the \c data and call the \c write in the \c writer with \c data as an argument.
59+
* The \c write in the \c writer will call \c get_payload which, instead of saving a new chunk of memory for data it
60+
* already contains, will return the \c data it had saved at the time of writing.
61+
*
62+
* @warning This method locks the \c mutex to avoid retrieving the \c data saved in another call to \c write.
63+
*
64+
* Thread safe.
65+
*
66+
* @param writer the writer who has to write the \c data.
67+
* @param data the data to be written by the \c writer.
68+
*
69+
* @return true if everything OK
70+
* @return false if something went wrong
71+
*/
72+
DDSPIPE_CORE_DllAPI
73+
bool write(
74+
fastdds::dds::DataWriter* writer,
75+
types::RtpsPayloadData* data);
76+
77+
/**
78+
* @brief write \c data with a \c writer with \c params but avoiding copying the \c data from and to the same
79+
* \c PayloadPool.
80+
*
81+
* Save the \c data and call the \c write in the \c writer with \c data and \c params as arguments.
82+
* The \c write in the \c writer will call \c get_payload which, instead of saving a new chunk of memory for data it
83+
* already contains, will return the \c data it had saved at the time of writing.
84+
*
85+
* @warning This method locks the \c mutex to avoid retrieving the \c data saved in another call to \c write.
86+
*
87+
* Thread safe.
88+
*
89+
* @param writer the writer who has to write the \c data.
90+
* @param data the data to be written by the \c writer.
91+
* @param params the data to be written by the \c writer.
92+
*
93+
* @return true if everything OK
94+
* @return false if something went wrong
95+
*/
96+
DDSPIPE_CORE_DllAPI
97+
bool write(
98+
fastdds::dds::DataWriter* writer,
99+
types::RtpsPayloadData* data,
100+
fastrtps::rtps::WriteParams& params);
101+
102+
/**
103+
* @brief write \c data with a \c writer with a \c handle but avoiding copying the \c data from and to the same
104+
* \c PayloadPool.
105+
*
106+
* Save the \c data and call the \c write in the \c writer with \c data and \c params as arguments.
107+
* The \c write in the \c writer will call \c get_payload which, instead of saving a new chunk of memory for data it
108+
* already contains, will return the \c data it had saved at the time of writing.
109+
*
110+
* @warning This method locks the \c mutex to avoid retrieving the \c data saved in another call to \c write.
111+
*
112+
* Thread safe.
113+
*
114+
* @param writer the writer who has to write the \c data.
115+
* @param data the data to be written by the \c writer.
116+
* @param handle the data to be used by the \c writer.
117+
*
118+
* @return true if everything OK
119+
* @return false if something went wrong
120+
*/
121+
DDSPIPE_CORE_DllAPI
122+
utils::ReturnCode write(
123+
fastdds::dds::DataWriter* writer,
124+
types::RtpsPayloadData* data,
125+
const fastrtps::rtps::InstanceHandle_t& handle);
126+
127+
/**
128+
* Instead of reserving a block of memory of \c size in the \c payload_pool, we can redirect the call to
129+
* \c get_payload providing the \c payload (that we saved in the call to \c write) and the \c payload_pool.
130+
*
131+
* @param size size of the new chunk of data to allocate in the \c payload_pool.
132+
* @param cache_change object to store the new data in.
133+
*
134+
* @return true if everything OK
135+
* @return false if something went wrong
136+
*/
137+
DDSPIPE_CORE_DllAPI
138+
virtual bool get_payload(
139+
uint32_t size,
140+
fastrtps::rtps::CacheChange_t& cache_change) override;
141+
142+
/**
143+
* @brief redirect the call to the \c release_payload in the \c payload_pool.
144+
*
145+
* @param data the data to get from the \c PayloadPool.
146+
* @param data_owner the \c PayloadPool to get the \c data from.
147+
* @param cache_change object to store the \c data in.
148+
*
149+
* @return true if everything OK
150+
* @return false if something went wrong
151+
*/
152+
DDSPIPE_CORE_DllAPI
153+
virtual bool get_payload(
154+
fastrtps::rtps::SerializedPayload_t& data,
155+
fastrtps::rtps::IPayloadPool*& data_owner,
156+
fastrtps::rtps::CacheChange_t& cache_change) override;
157+
158+
/**
159+
* @brief redirect the call to the \c release_payload in the \c payload_pool.
160+
* @param cache_change object to release.
161+
*
162+
* @return true if everything OK
163+
* @return false if something went wrong
164+
*/
165+
DDSPIPE_CORE_DllAPI
166+
virtual bool release_payload(
167+
fastrtps::rtps::CacheChange_t& cache_change) override;
168+
169+
protected:
170+
171+
//! Mutex to prevent simultaneous calls to write
172+
std::mutex mutex_;
173+
174+
//! A pointer to the payload_ we want to write down in the PayloadPool.
175+
types::Payload* payload_;
176+
177+
//! The \c PayloadPool the \c PayloadPoolMediator is mediating for.
178+
const std::shared_ptr<PayloadPool>& payload_pool_;
179+
};
180+
181+
} /* namespace core */
182+
} /* namespace ddspipe */
183+
} /* namespace eprosima */

‎ddspipe_core/src/cpp/efficiency/payload/FastPayloadPool.cpp‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -34,12 +34,7 @@ bool FastPayloadPool::get_payload(
3434
Payload& payload)
3535
{
3636
// Reserve new payload
37-
if (!reserve_(size, payload))
38-
{
39-
return false;
40-
}
41-
42-
return true;
37+
return reserve_(size, payload);
4338
}
4439

4540
bool FastPayloadPool::get_payload(
@@ -50,6 +45,8 @@ bool FastPayloadPool::get_payload(
5045
// If we are not the owner, create a new payload. Else, reference the existing one
5146
if (data_owner != this)
5247
{
48+
logDebug(DDSPIPE_PAYLOADPOOL_FAST, "Copying payload with ptr: " << static_cast<void*>(src_payload.data) << ".");
49+
5350
// Store space for payload
5451
if (!get_payload(src_payload.max_size, target_payload))
5552
{
@@ -62,6 +59,9 @@ bool FastPayloadPool::get_payload(
6259
}
6360
else
6461
{
62+
logDebug(DDSPIPE_PAYLOADPOOL_FAST,
63+
"Referencing payload with ptr: " << static_cast<void*>(src_payload.data) << ".");
64+
6565
// IMPORTANT: If payload has been reserved from this object, it must have the +4 bytes before data
6666
// We get this value and add a +1 to set that is referencing from one more payload
6767
MetaInfoType* reference_place = reinterpret_cast<MetaInfoType*>(src_payload.data);
@@ -116,7 +116,6 @@ bool FastPayloadPool::reserve_(
116116
return false;
117117
}
118118

119-
120119
// Allocate memory + 4 bytes for reference
121120
void* memory_allocated = std::malloc(size + sizeof(MetaInfoType));
122121

‎ddspipe_core/src/cpp/efficiency/payload/PayloadPoolMediator.cpp‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
// Copyright 2023 Proyectos y Sistemas de Mantenimiento SL (eProsima).
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
/**
16+
* @file PayloadPoolMediator.cpp
17+
*
18+
*/
19+
20+
#include <cpp_utils/exception/UnsupportedException.hpp>
21+
#include <cpp_utils/exception/InconsistencyException.hpp>
22+
#include <cpp_utils/Log.hpp>
23+
24+
#include <ddspipe_core/efficiency/payload/PayloadPoolMediator.hpp>
25+
26+
namespace eprosima {
27+
namespace ddspipe {
28+
namespace core {
29+
30+
PayloadPoolMediator::PayloadPoolMediator(
31+
const std::shared_ptr<PayloadPool>& payload_pool)
32+
: payload_pool_(payload_pool)
33+
{
34+
}
35+
36+
bool PayloadPoolMediator::write(
37+
fastdds::dds::DataWriter* writer,
38+
types::RtpsPayloadData* data)
39+
{
40+
// Lock the mutex_ to ensure that the payload hasn't changed when we retrieve it in get_payload.
41+
std::lock_guard<std::mutex> lock(mutex_);
42+
43+
payload_ = &data->payload;
44+
45+
return writer->write(data);
46+
}
47+
48+
bool PayloadPoolMediator::write(
49+
fastdds::dds::DataWriter* writer,
50+
types::RtpsPayloadData* data,
51+
fastrtps::rtps::WriteParams& params)
52+
{
53+
// Lock the mutex_ to ensure that the payload hasn't changed when we retrieve it in get_payload.
54+
std::lock_guard<std::mutex> lock(mutex_);
55+
56+
payload_ = &data->payload;
57+
58+
return writer->write(data, params);
59+
}
60+
61+
utils::ReturnCode PayloadPoolMediator::write(
62+
fastdds::dds::DataWriter* writer,
63+
types::RtpsPayloadData* data,
64+
const fastrtps::rtps::InstanceHandle_t& handle)
65+
{
66+
// Lock the mutex_ to ensure that the payload hasn't changed when we retrieve it in get_payload.
67+
std::lock_guard<std::mutex> lock(mutex_);
68+
69+
payload_ = &data->payload;
70+
71+
return writer->write(data, handle);
72+
}
73+
74+
bool PayloadPoolMediator::get_payload(
75+
uint32_t size,
76+
fastrtps::rtps::CacheChange_t& cache_change)
77+
{
78+
fastrtps::rtps::IPayloadPool* payload_owner{payload_pool_.get()};
79+
80+
return get_payload(*payload_, payload_owner, cache_change);
81+
}
82+
83+
bool PayloadPoolMediator::get_payload(
84+
fastrtps::rtps::SerializedPayload_t& data,
85+
fastrtps::rtps::IPayloadPool*& data_owner,
86+
fastrtps::rtps::CacheChange_t& cache_change)
87+
{
88+
return payload_pool_->get_payload(data, data_owner, cache_change);
89+
}
90+
91+
bool PayloadPoolMediator::release_payload(
92+
fastrtps::rtps::CacheChange_t& cache_change)
93+
{
94+
return payload_pool_->release_payload(cache_change);
95+
}
96+
97+
} /* namespace core */
98+
} /* namespace ddspipe */
99+
} /* namespace eprosima */

‎ddspipe_participants/include/ddspipe_participants/writer/dds/CommonWriter.hpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
2323
#include <fastdds/dds/topic/Topic.hpp>
2424

25+
#include <ddspipe_core/efficiency/payload/PayloadPoolMediator.hpp>
2526
#include <ddspipe_core/types/participant/ParticipantId.hpp>
2627
#include <ddspipe_core/types/data/RtpsPayloadData.hpp>
2728
#include <ddspipe_core/types/topic/dds/DdsTopic.hpp>
@@ -146,7 +147,7 @@ class CommonWriter : public BaseWriter
146147
// INTERNAL VARIABLES
147148
/////////////////////////
148149

149-
const std::shared_ptr<core::PayloadPool>& payload_pool_;
150+
std::shared_ptr<core::PayloadPoolMediator> payload_pool_;
150151

151152
core::types::DdsTopic topic_;
152153

‎ddspipe_participants/src/cpp/reader/dds/CommonReader.cpp‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,9 @@ void CommonReader::init()
7272
reader_ = dds_subscriber_->create_datareader(
7373
dds_topic_,
7474
reckon_reader_qos_(),
75-
nullptr);
75+
nullptr,
76+
eprosima::fastdds::dds::StatusMask::all(),
77+
payload_pool_);
7678

7779
if (!reader_)
7880
{
@@ -126,8 +128,7 @@ utils::ReturnCode CommonReader::take_nts_(
126128
return utils::ReturnCode::RETCODE_NO_DATA;
127129
}
128130

129-
RtpsPayloadData* rtps_data = new core::types::RtpsPayloadData();
130-
data.reset(rtps_data);
131+
RtpsPayloadData* rtps_data;
131132
fastdds::dds::SampleInfo info;
132133

133134
// Loop for read messages until we receive one that does not come from same participant
@@ -136,8 +137,15 @@ utils::ReturnCode CommonReader::take_nts_(
136137
// 2. ignore_local_endpoints would be override by xml configuration
137138
while (true)
138139
{
140+
// Ensure that the previous Payload gets destroyed to avoid memory leaks.
141+
rtps_data = new core::types::RtpsPayloadData();
142+
data.reset(rtps_data);
143+
139144
auto ret = reader_->take_next_sample(rtps_data, &info);
140145

146+
// Save the payload owner so that the memory is freed correctly.
147+
rtps_data->payload_owner = payload_pool_.get();
148+
141149
// If error reading data
142150
if (!ret)
143151
{
@@ -247,8 +255,6 @@ void CommonReader::fill_received_data_(
247255
// Get Participant receiver
248256
data_to_fill.participant_receiver = participant_id_;
249257

250-
// TODO modify when access to payload pool
251-
252258
// Set Instance Handle to data_to_fill
253259
if (topic_.topic_qos.keyed)
254260
{

‎ddspipe_participants/src/cpp/reader/rtps/CommonReader.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,7 @@ void CommonReader::fill_received_data_(
215215
{
216216
eprosima::fastrtps::rtps::IPayloadPool* payload_owner =
217217
const_cast<eprosima::fastrtps::rtps::IPayloadPool*>(received_change.payload_owner());
218+
218219
payload_pool_->get_payload(
219220
received_change.serializedPayload,
220221
payload_owner,

0 commit comments

Comments
 (0)