Skip to content

Commit d3ca40c

Browse files
RTPS WriterHistory refactor (#4966)
* Refs #21082. Remove templated version of `new_change`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Call simple version of `new_change` in rtps examples. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Call simple version of `new_change` in `TypeLookupManager`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on EDP. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on PDP. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on SecurityManager. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on unit tests. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on WLP. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on MonitorService. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on blackbox tests. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Fix blackbox tests. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `History::do_reserve_cache`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `new_change` overload receiving `std::function`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `is_pool_initialized`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. WriterHistory keeps a change pool. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Implementation of `release_change` moved to WriterHistory. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `release_change` from `RTPSWriter`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Add `create_change` to `WriterHistory`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Change calls from `new_change` to `create_change`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `new_change` from `RTPSWriter`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove `remove_older_changes` from `RTPSWriter`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Add payload pool to `WriterHistory`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Add new `create_change` overload. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Using new `create_change` overload where relevant. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on `IPersistenceService`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor on `DataWriterHistory`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Several methods to create writers removed. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Move data sharing pool initialization to DataWriterImpl. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor creation of rtps writer. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Refactor PersistentWriter. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove constructors taking pools. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. RTPSWriter does not handle pools. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Add pool getters to WriterHistory. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Remove pool references on StatexxxWriter. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Move pools from `Endpoint` to `BaseReader`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Avoid accessing history attributes. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Fixes on SecurityManager. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Fix build after rebase. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Please linters. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Fix release order in `WriterHistory` destructor. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Move datasharing pool initialization to `RTPSWriter`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21082. Fix negative tests. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21137. Apply review suggestions. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21137. Apply suggestions from code review Signed-off-by: Miguel Company <miguelcompany@eprosima.com> Co-authored-by: Eduardo Ponz Segrelles <eduardoponz@eprosima.com> * Refs #21137. Add note to versions.md Signed-off-by: eduponz <eduardoponz@eprosima.com> --------- Signed-off-by: Miguel Company <miguelcompany@eprosima.com> Signed-off-by: eduponz <eduardoponz@eprosima.com> Co-authored-by: Eduardo Ponz Segrelles <eduardoponz@eprosima.com>
1 parent 4c9f1d6 commit d3ca40c

77 files changed

Lines changed: 867 additions & 1307 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

examples/cpp/rtps/AsSocket/TestWriterSocket.cpp

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -89,10 +89,7 @@ void TestWriterSocket::run(
8989
{
9090
for (int i = 0; i < nmsgs; ++i )
9191
{
92-
CacheChange_t* ch = mp_writer->new_change([]() -> uint32_t
93-
{
94-
return 255;
95-
}, ALIVE);
92+
CacheChange_t* ch = mp_history->create_change(255, ALIVE);
9693
#if defined(_WIN32)
9794
ch->serializedPayload.length =
9895
sprintf_s((char*)ch->serializedPayload.data, 255, "My example string %d", i) + 1;

examples/cpp/rtps/Persistent/TestWriterPersistent.cpp

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -112,18 +112,12 @@ void TestWriterPersistent::run(
112112

113113
for (int i = 0; i < samples; ++i )
114114
{
115-
CacheChange_t* ch = mp_writer->new_change([]() -> uint32_t
116-
{
117-
return 255;
118-
}, ALIVE);
115+
CacheChange_t* ch = mp_history->create_change(255, ALIVE);
119116
if (!ch) // In the case history is full, remove some old changes
120117
{
121118
std::cout << "cleaning history...";
122-
mp_writer->remove_older_changes(20);
123-
ch = mp_writer->new_change([]() -> uint32_t
124-
{
125-
return 255;
126-
}, ALIVE);
119+
mp_history->remove_min_change();
120+
ch = mp_history->create_change(255, ALIVE);
127121
}
128122

129123
#if defined(_WIN32)

examples/cpp/rtps/Registered/TestWriterRegistered.cpp

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -103,18 +103,12 @@ void TestWriterRegistered::run(
103103

104104
for (int i = 0; i < samples; ++i )
105105
{
106-
CacheChange_t* ch = mp_writer->new_change([]() -> uint32_t
107-
{
108-
return 255;
109-
}, ALIVE);
106+
CacheChange_t* ch = mp_history->create_change(255, ALIVE);
110107
if (!ch) // In the case history is full, remove some old changes
111108
{
112109
std::cout << "cleaning history...";
113-
mp_writer->remove_older_changes(20);
114-
ch = mp_writer->new_change([]() -> uint32_t
115-
{
116-
return 255;
117-
}, ALIVE);
110+
mp_history->remove_min_change();
111+
ch = mp_history->create_change(255, ALIVE);
118112
}
119113

120114
#if defined(_WIN32)

include/fastdds/rtps/Endpoint.hpp

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -62,11 +62,6 @@ class Endpoint
6262

6363
virtual ~Endpoint()
6464
{
65-
// As releasing the change pool will delete the cache changes it owns,
66-
// the payload pool may be called to release their payloads, so we should
67-
// ensure that the payload pool is destroyed after the change pool.
68-
change_pool_.reset();
69-
payload_pool_.reset();
7065
}
7166

7267
public:
@@ -120,12 +115,6 @@ class Endpoint
120115
//!Endpoint Mutex
121116
mutable RecursiveTimedMutex mp_mutex;
122117

123-
//!Pool of serialized payloads.
124-
std::shared_ptr<IPayloadPool> payload_pool_;
125-
126-
//!Pool of cache changes.
127-
std::shared_ptr<IChangePool> change_pool_;
128-
129118
//!Fixed size of payloads
130119
uint32_t fixed_payload_size_ = 0;
131120

include/fastdds/rtps/RTPSDomain.hpp

Lines changed: 11 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -113,92 +113,32 @@ class RTPSDomain
113113

114114
/**
115115
* Create a RTPSWriter in a participant.
116-
* @param p Pointer to the RTPSParticipant.
117-
* @param watt Writer Attributes.
118-
* @param hist Pointer to the WriterHistory.
119-
* @param listen Pointer to the WriterListener.
120-
* @return Pointer to the created RTPSWriter.
121116
*
122-
* \warning The returned pointer is invalidated after a call to removeRTPSWriter() or stopAll(),
123-
* so its use may result in undefined behaviour.
124-
*/
125-
FASTDDS_EXPORTED_API static RTPSWriter* createRTPSWriter(
126-
RTPSParticipant* p,
127-
WriterAttributes& watt,
128-
WriterHistory* hist,
129-
WriterListener* listen = nullptr);
130-
131-
/**
132-
* Create a RTPSWriter in a participant using a custom payload pool.
133-
* @param p Pointer to the RTPSParticipant.
134-
* @param watt Writer Attributes.
135-
* @param payload_pool Shared pointer to the IPayloadPool
136-
* @param hist Pointer to the WriterHistory.
137-
* @param listen Pointer to the WriterListener.
138-
* @return Pointer to the created RTPSWriter.
139-
*
140-
* \warning The returned pointer is invalidated after a call to removeRTPSWriter() or stopAll(),
141-
* so its use may result in undefined behaviour.
142-
*/
143-
FASTDDS_EXPORTED_API static RTPSWriter* createRTPSWriter(
144-
RTPSParticipant* p,
145-
WriterAttributes& watt,
146-
const std::shared_ptr<IPayloadPool>& payload_pool,
147-
WriterHistory* hist,
148-
WriterListener* listen = nullptr);
149-
150-
/**
151-
* Create a RTPSWriter in a participant using a custom payload pool.
152-
* @param p Pointer to the RTPSParticipant.
153-
* @param watt Writer Attributes.
154-
* @param payload_pool Shared pointer to the IPayloadPool
155-
* @param change_pool Shared pointer to the IChangePool
156-
* @param hist Pointer to the WriterHistory.
157-
* @param listen Pointer to the WriterListener.
158-
* @return Pointer to the created RTPSWriter.
117+
* @param p Pointer to the RTPSParticipant.
118+
* @param watt Writer Attributes.
119+
* @param hist Pointer to the WriterHistory.
120+
* @param listen Pointer to the WriterListener.
159121
*
160-
* \warning The returned pointer is invalidated after a call to removeRTPSWriter() or stopAll(),
161-
* so its use may result in undefined behaviour.
162-
*/
163-
FASTDDS_EXPORTED_API static RTPSWriter* createRTPSWriter(
164-
RTPSParticipant* p,
165-
WriterAttributes& watt,
166-
const std::shared_ptr<IPayloadPool>& payload_pool,
167-
const std::shared_ptr<IChangePool>& change_pool,
168-
WriterHistory* hist,
169-
WriterListener* listen = nullptr);
170-
171-
/**
172-
* Create a RTPSWriter in a participant using a custom payload pool.
173-
* @param p Pointer to the RTPSParticipant.
174-
* @param entity_id Specific entity id to use for the created writer.
175-
* @param watt Writer Attributes.
176-
* @param payload_pool Shared pointer to the IPayloadPool
177-
* @param change_pool Shared pointer to the IChangePool
178-
* @param hist Pointer to the WriterHistory.
179-
* @param listen Pointer to the WriterListener.
180122
* @return Pointer to the created RTPSWriter.
181123
*
182124
* \warning The returned pointer is invalidated after a call to removeRTPSWriter() or stopAll(),
183125
* so its use may result in undefined behaviour.
184126
*/
185127
FASTDDS_EXPORTED_API static RTPSWriter* createRTPSWriter(
186128
RTPSParticipant* p,
187-
const EntityId_t& entity_id,
188129
WriterAttributes& watt,
189-
const std::shared_ptr<IPayloadPool>& payload_pool,
190-
const std::shared_ptr<IChangePool>& change_pool,
191130
WriterHistory* hist,
192131
WriterListener* listen = nullptr);
193132

194133
/**
195134
* Create a RTPSWriter in a participant.
196-
* @param p Pointer to the RTPSParticipant.
197-
* @param entity_id Specific entity id to use for the created writer.
198-
* @param watt Writer Attributes.
199-
* @param payload_pool Shared pointer to the IPayloadPool
200-
* @param hist Pointer to the WriterHistory.
201-
* @param listen Pointer to the WriterListener.
135+
*
136+
* @param p Pointer to the RTPSParticipant.
137+
* @param entity_id Specific entity id to use for the created writer.
138+
* @param watt Writer Attributes.
139+
* @param hist Pointer to the WriterHistory.
140+
* @param listen Pointer to the WriterListener.
141+
*
202142
* @return Pointer to the created RTPSWriter.
203143
*
204144
* \warning The returned pointer is invalidated after a call to removeRTPSWriter() or stopAll(),
@@ -208,7 +148,6 @@ class RTPSDomain
208148
RTPSParticipant* p,
209149
const EntityId_t& entity_id,
210150
WriterAttributes& watt,
211-
const std::shared_ptr<IPayloadPool>& payload_pool,
212151
WriterHistory* hist,
213152
WriterListener* listen = nullptr);
214153

include/fastdds/rtps/history/History.hpp

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -271,10 +271,6 @@ class History
271271
//!Print the seqNum of the changes in the History (for debuggisi, mng purposes).
272272
void print_changes_seqNum2();
273273

274-
FASTDDS_EXPORTED_API virtual bool do_reserve_cache(
275-
CacheChange_t** change,
276-
uint32_t size) = 0;
277-
278274
FASTDDS_EXPORTED_API virtual void do_release_cache(
279275
CacheChange_t* ch) = 0;
280276

include/fastdds/rtps/history/ReaderHistory.hpp

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -233,10 +233,6 @@ class ReaderHistory : public History
233233

234234
protected:
235235

236-
FASTDDS_EXPORTED_API bool do_reserve_cache(
237-
CacheChange_t** change,
238-
uint32_t size) override;
239-
240236
FASTDDS_EXPORTED_API void do_release_cache(
241237
CacheChange_t* ch) override;
242238

include/fastdds/rtps/history/WriterHistory.hpp

Lines changed: 106 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,23 @@
2020
#ifndef FASTDDS_RTPS_HISTORY__WRITERHISTORY_HPP
2121
#define FASTDDS_RTPS_HISTORY__WRITERHISTORY_HPP
2222

23-
#include <fastdds/rtps/history/History.hpp>
23+
#include <cstdint>
24+
#include <memory>
25+
26+
#include <fastdds/fastdds_dll.hpp>
2427
#include <fastdds/dds/log/Log.hpp>
28+
#include <fastdds/rtps/common/CacheChange.hpp>
29+
#include <fastdds/rtps/common/ChangeKind_t.hpp>
30+
#include <fastdds/rtps/common/InstanceHandle.hpp>
31+
#include <fastdds/rtps/history/History.hpp>
32+
#include <fastdds/rtps/history/IChangePool.hpp>
33+
#include <fastdds/rtps/history/IPayloadPool.hpp>
2534

2635
namespace eprosima {
2736
namespace fastdds {
2837
namespace rtps {
2938

39+
class HistoryAttributes;
3040
class RTPSWriter;
3141
class WriteParams;
3242

@@ -48,12 +58,81 @@ class WriterHistory : public rtps::History
4858
public:
4959

5060
/**
51-
* Constructor of the WriterHistory.
61+
* @brief Construct a WriterHistory.
62+
*
63+
* @param att Attributes configuring the WriterHistory.
64+
*/
65+
FASTDDS_EXPORTED_API WriterHistory(
66+
const HistoryAttributes& att);
67+
68+
/**
69+
* @brief Construct a WriterHistory with a custom payload pool.
70+
*
71+
* @param att Attributes configuring the WriterHistory.
72+
* @param payload_pool Pool of payloads to be used by the WriterHistory.
73+
*/
74+
FASTDDS_EXPORTED_API WriterHistory(
75+
const HistoryAttributes& att,
76+
const std::shared_ptr<IPayloadPool>& payload_pool);
77+
78+
/**
79+
* @brief Construct a WriterHistory with custom payload and change pools.
80+
*
81+
* @param att Attributes configuring the WriterHistory.
82+
* @param payload_pool Pool of payloads to be used by the WriterHistory.
83+
* @param change_pool Pool of changes to be used by the WriterHistory.
5284
*/
5385
FASTDDS_EXPORTED_API WriterHistory(
54-
const HistoryAttributes& att);
86+
const HistoryAttributes& att,
87+
const std::shared_ptr<IPayloadPool>& payload_pool,
88+
const std::shared_ptr<IChangePool>& change_pool);
89+
5590
FASTDDS_EXPORTED_API virtual ~WriterHistory() override;
5691

92+
/**
93+
* @brief Get the payload pool used by this history.
94+
*
95+
* @return Reference to the payload pool used by this history.
96+
*/
97+
FASTDDS_EXPORTED_API const std::shared_ptr<IPayloadPool>& get_payload_pool() const;
98+
99+
/**
100+
* @brief Get the change pool used by this history.
101+
*
102+
* @return Reference to the change pool used by this history.
103+
*/
104+
FASTDDS_EXPORTED_API const std::shared_ptr<IChangePool>& get_change_pool() const;
105+
106+
/**
107+
* @brief Create a new CacheChange_t object.
108+
*
109+
* @param change_kind Kind of the change.
110+
* @param handle InstanceHandle_t of the change.
111+
*
112+
* @return Pointer to the new CacheChange_t object.
113+
*
114+
* @pre A writer has been associated with this history
115+
*/
116+
FASTDDS_EXPORTED_API CacheChange_t* create_change(
117+
ChangeKind_t change_kind,
118+
InstanceHandle_t handle = c_InstanceHandle_Unknown);
119+
120+
/**
121+
* @brief Create a new CacheChange_t object with a specific payload size.
122+
*
123+
* @param payload_size Size of the payload.
124+
* @param change_kind Kind of the change.
125+
* @param handle InstanceHandle_t of the change.
126+
*
127+
* @return Pointer to the new CacheChange_t object.
128+
*
129+
* @pre A writer has been associated with this history
130+
*/
131+
FASTDDS_EXPORTED_API CacheChange_t* create_change(
132+
uint32_t payload_size,
133+
ChangeKind_t change_kind,
134+
InstanceHandle_t handle = c_InstanceHandle_Unknown);
135+
57136
/**
58137
* Add a CacheChange_t to the WriterHistory.
59138
* @param a_change Pointer to the CacheChange_t to be added.
@@ -141,11 +220,24 @@ class WriterHistory : public rtps::History
141220
return m_lastCacheChangeSeqNum + 1;
142221
}
143222

144-
protected:
223+
/**
224+
* Release a change when it is not being used anymore.
225+
*
226+
* @param ch Pointer to the cache change to be released.
227+
*
228+
* @returns whether the operation succeeded or not
229+
*
230+
* @pre
231+
* @li A writer has been associated with this history
232+
* @li @c ch is not @c nullptr
233+
* @li @c ch points to a cache change obtained from a call to @c this->create_change
234+
*
235+
* @post memory pointed to by @c ch is not accessed
236+
*/
237+
FASTDDS_EXPORTED_API bool release_change(
238+
CacheChange_t* ch);
145239

146-
FASTDDS_EXPORTED_API bool do_reserve_cache(
147-
CacheChange_t** change,
148-
uint32_t size) override;
240+
protected:
149241

150242
FASTDDS_EXPORTED_API void do_release_cache(
151243
CacheChange_t* ch) override;
@@ -211,9 +303,9 @@ class WriterHistory : public rtps::History
211303
}
212304

213305
//!Last CacheChange Sequence Number added to the History.
214-
SequenceNumber_t m_lastCacheChangeSeqNum;
306+
SequenceNumber_t m_lastCacheChangeSeqNum {};
215307
//!Pointer to the associated RTPSWriter;
216-
RTPSWriter* mp_writer;
308+
RTPSWriter* mp_writer = nullptr;
217309

218310
uint32_t high_mark_for_frag_ = 0;
219311

@@ -248,6 +340,11 @@ class WriterHistory : public rtps::History
248340

249341
void set_fragments(
250342
CacheChange_t* change);
343+
344+
/// Reference to the change pool used by this history.
345+
std::shared_ptr<IChangePool> change_pool_;
346+
/// Reference to the payload pool used by this history.
347+
std::shared_ptr<IPayloadPool> payload_pool_;
251348
};
252349

253350
} // namespace rtps

0 commit comments

Comments
 (0)