Skip to content

Commit c808afe

Browse files
emiliocuestafemiliocuestaf
authored
Forward unregister and dispose messages (#172)
* Add code to forward unregister and dispose messages Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Add payload mediator as its required Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Uncrustify Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Change on autodispose QoS policy on writer Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Fix wrong comparison of return codes Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Remove unnecessary () since devutils will be fixed Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Remove unwanted () in ret codes Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Apply revision Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Uncrustify Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> --------- Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com>
1 parent 0032cdd commit c808afe

4 files changed

Lines changed: 110 additions & 14 deletions

File tree

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

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,48 @@ class PayloadPoolMediator : public fastdds::rtps::IPayloadPool
121121
types::RtpsPayloadData* data,
122122
const fastdds::rtps::InstanceHandle_t& handle);
123123

124+
/**
125+
* @brief dispose \c data with a \c writer avoiding copying the \c data from and to the same \c PayloadPool.
126+
*
127+
* Save the \c data and call the \c dispose in the \c writer with \c data as an argument.
128+
*
129+
* @warning This method locks the \c mutex to avoid retrieving the \c data saved in another call to \c write.
130+
*
131+
* Thread safe.
132+
*
133+
* @param writer the writer who has to dispose the \c data.
134+
* @param data the data to be disposed by the \c writer.
135+
* @param handle the instance handle to be used by the \c writer.
136+
*
137+
* @return ReturnCode::OK if the instance is correctly disposed or a ReturnCode related to the specific error otherwise.
138+
*/
139+
DDSPIPE_CORE_DllAPI
140+
utils::ReturnCode dispose(
141+
fastdds::dds::DataWriter* writer,
142+
types::RtpsPayloadData* data,
143+
const fastdds::rtps::InstanceHandle_t& handle);
144+
145+
/**
146+
* @brief unregister \c data with a \c writer avoiding copying the \c data from and to the same \c PayloadPool.
147+
*
148+
* Save the \c data and call the \c unregister_instance in the \c writer with \c data as an argument.
149+
*
150+
* @warning This method locks the \c mutex to avoid retrieving the \c data saved in another call to \c write.
151+
*
152+
* Thread safe.
153+
*
154+
* @param writer the writer who has to unregister the \c data.
155+
* @param data the data to be unregistered by the \c writer.
156+
* @param handle the instance handle to be used by the \c writer.
157+
*
158+
* @return ReturnCode::OK if the instance is correctly unregistered or a ReturnCode related to the specific error otherwise.
159+
*/
160+
DDSPIPE_CORE_DllAPI
161+
utils::ReturnCode unregister_instance(
162+
fastdds::dds::DataWriter* writer,
163+
types::RtpsPayloadData* data,
164+
const fastdds::rtps::InstanceHandle_t& handle);
165+
124166
/**
125167
* Instead of reserving a block of memory of \c size in the \c payload_pool, we can redirect the call to
126168
* \c get_payload providing the \c payload (that we saved in the call to \c write) and the \c payload_pool.

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,32 @@ utils::ReturnCode PayloadPoolMediator::write(
7171
return writer->write(data, handle);
7272
}
7373

74+
utils::ReturnCode PayloadPoolMediator::dispose(
75+
fastdds::dds::DataWriter* writer,
76+
types::RtpsPayloadData* data,
77+
const fastdds::rtps::InstanceHandle_t& handle)
78+
{
79+
// Lock the mutex_ to ensure that the payload hasn't changed when we retrieve it in get_payload.
80+
std::lock_guard<std::mutex> lock(mutex_);
81+
82+
payload_ = &data->payload;
83+
84+
return writer->dispose(data, handle);
85+
}
86+
87+
utils::ReturnCode PayloadPoolMediator::unregister_instance(
88+
fastdds::dds::DataWriter* writer,
89+
types::RtpsPayloadData* data,
90+
const fastdds::rtps::InstanceHandle_t& handle)
91+
{
92+
// Lock the mutex_ to ensure that the payload hasn't changed when we retrieve it in get_payload.
93+
std::lock_guard<std::mutex> lock(mutex_);
94+
95+
payload_ = &data->payload;
96+
97+
return writer->unregister_instance(data, handle);
98+
}
99+
74100
bool PayloadPoolMediator::get_payload(
75101
uint32_t size,
76102
eprosima::fastdds::rtps::SerializedPayload_t& payload)

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

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -215,11 +215,6 @@ utils::ReturnCode CommonReader::take_nts_(
215215

216216
EPROSIMA_LOG_INFO(DDSPIPE_DDS_READER, "Taking data in " << participant_id_ << " for topic " << topic_ << ".");
217217

218-
// Check if there is data available
219-
if (!(reader_->get_unread_count() > 0))
220-
{
221-
return utils::ReturnCode::RETCODE_NO_DATA;
222-
}
223218

224219
std::unique_ptr<RtpsPayloadData> rtps_data;
225220
fastdds::dds::SampleInfo info;
@@ -375,7 +370,15 @@ void CommonReader::fill_received_data_(
375370
data_to_fill.kind = ChangeKind::NOT_ALIVE_DISPOSED;
376371
break;
377372

373+
case fastdds::dds::InstanceStateKind::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE:
374+
data_to_fill.kind = ChangeKind::NOT_ALIVE_UNREGISTERED;
375+
break;
376+
378377
default:
378+
EPROSIMA_LOG_WARNING(DDSPIPE_DDS_READER,
379+
"Received unknown InstanceStateKind from Fast DDS: " << static_cast<int>(info.instance_state)
380+
<<
381+
". Setting ChangeKind to NOT_ALIVE_UNREGISTERED by default.");
379382
data_to_fill.kind = ChangeKind::NOT_ALIVE_UNREGISTERED;
380383
break;
381384
}

‎ddspipe_participants/src/cpp/writer/dds/CommonWriter.cpp‎

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -162,15 +162,35 @@ utils::ReturnCode CommonWriter::write_nts_(
162162
return utils::ReturnCode::RETCODE_ERROR;
163163
}
164164

165-
// WARNING: At the time of this writing, there is no DataWriter API to write both with parameters and instance handle.
166-
// However, the current implementation of write without instance handle always computes its value via type support
167-
// (if it corresponds to a keyed topic), so it is equivalent to write with instance handle (and can hence use the
168-
// write with params overload to cover all cases). Future developers should be aware of this and might need to
169-
// update this method if the DataWriter implementation changes at some point.
170-
return payload_pool_->write(writer_, &rtps_data, wparams);
171-
172-
// TODO: handle dipose case -> DataWriter::write will always send ALIVE changes, so this case must be handled
173-
// with additional logic (e.g. by using unregister_instance instead of write).
165+
// NOTE: Currently dispose and unregister messages are being sent through the payload pool. This could unnecessary
166+
// slow if they don't have payloads attached, but it is left this way in order to ensure compatibility with other
167+
// vendors and the standard.
168+
169+
switch (rtps_data.kind){
170+
case core::types::ChangeKind::ALIVE:
171+
// WARNING: At the time of this writing, there is no DataWriter API to write both with parameters and instance handle.
172+
// However, the current implementation of write without instance handle always computes its value via type support
173+
// (if it corresponds to a keyed topic), so it is equivalent to write with instance handle (and can hence use the
174+
// write with params overload to cover all cases). Future developers should be aware of this and might need to
175+
// update this method if the DataWriter implementation changes at some point.
176+
return payload_pool_->write(writer_, &rtps_data, wparams);
177+
case core::types::ChangeKind::NOT_ALIVE_DISPOSED:
178+
return payload_pool_->dispose(writer_, &rtps_data, rtps_data.instanceHandle);
179+
case core::types::ChangeKind::NOT_ALIVE_UNREGISTERED:
180+
return payload_pool_->unregister_instance(writer_, &rtps_data, rtps_data.instanceHandle);
181+
case core::types::ChangeKind::NOT_ALIVE_DISPOSED_UNREGISTERED:
182+
{
183+
utils::ReturnCode ret = payload_pool_->dispose(writer_, &rtps_data, rtps_data.instanceHandle);
184+
if (ret == utils::ReturnCode::RETCODE_OK)
185+
{
186+
return ret;
187+
}
188+
return payload_pool_->unregister_instance(writer_, &rtps_data, rtps_data.instanceHandle);
189+
}
190+
default:
191+
EPROSIMA_LOG_ERROR(DDSPIPE_DDS_WRITER, "Unknown RtpsPayloadData kind.");
192+
return utils::ReturnCode::RETCODE_ERROR;
193+
}
174194
}
175195

176196
void CommonWriter::update_partitions(
@@ -238,6 +258,11 @@ fastdds::dds::DataWriterQos CommonWriter::reckon_writer_qos_() const noexcept
238258
// Set minimum deadline so it matches with everything
239259
qos.deadline().period = eprosima::fastdds::dds::Duration_t(0);
240260

261+
// Setting this to false keeps the behavior of the original writer
262+
// If only _D (only dispose) was sent, the expected is that only _D (only dispose) is forwarded
263+
// If only U_ (only unregister) was sent, the expected is that only U_ (only unregister) is forwarded
264+
qos.writer_data_lifecycle().autodispose_unregistered_instances = false;
265+
241266
return qos;
242267
}
243268

0 commit comments

Comments
 (0)