Skip to content

Commit 4ef747b

Browse files
emiliocuestafcferreiragonz
emiliocuestaf
authored and
cferreiragonz
committed
Allow empty payloads in dispose/unregister operations (#6217)
* Allowing to send empty payloads in dispose/unregister ops Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Add tests Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Uncrustify Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Refactor to keep current execution path for most cases Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Making readers able to read empty paoload messages Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Adding tests for RTPS readers Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Fix trace logging in topic_instances example, that always showed the same color on dispose Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Uncrustify Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Make writer write empty payload only if get_payload fails Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Late disposes are not received in history and cause segfault when freed Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Fix asan problems Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Apply easy suggestions from code review Co-authored-by: Miguel Company <miguelcompany@eprosima.com> Signed-off-by: Emilio Cuesta Fernandez <emiliocuesta@eprosima.com> * Apply revision part2 Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Update src/cpp/fastdds/publisher/DataWriterImpl.cpp Co-authored-by: Miguel Company <miguelcompany@eprosima.com> Signed-off-by: Emilio Cuesta Fernandez <emiliocuesta@eprosima.com> * Replace old signature name Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Remove unnecessary check Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Adding assert in case that should not happen Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> * Removing line Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> --------- Signed-off-by: Emilio Cuesta <emiliocuesta@eprosima.com> Signed-off-by: Emilio Cuesta Fernandez <emiliocuesta@eprosima.com> Co-authored-by: Miguel Company <miguelcompany@eprosima.com> (cherry picked from commit 47bfac0) # Conflicts: # test/unittest/dds/publisher/DataWriterTests.cpp
1 parent aaf47ca commit 4ef747b

9 files changed

Lines changed: 946 additions & 16 deletions

File tree

‎examples/cpp/topic_instances/SubscriberApp.cpp‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,9 @@ void SubscriberApp::on_data_available(
162162
samples_per_instance_[info.instance_handle] = 1;
163163
}
164164

165+
// Store the color name in the instance
166+
color_per_instance_[info.instance_handle] = shape_.color();
167+
165168
// Print the shape
166169
std::cout << shape_.color() << " Shape with size " << shape_.shapesize()
167170
<< " at X:" << shape_.x() << ", Y:" << shape_.y() << " RECEIVED" << std::endl;
@@ -175,7 +178,9 @@ void SubscriberApp::on_data_available(
175178
}
176179
else if (info.instance_state == NOT_ALIVE_DISPOSED_INSTANCE_STATE)
177180
{
178-
std::cout << shape_.color() << " Shape has been disposed" << std::endl;
181+
std::string instance_name = shape_.color() !=
182+
"" ? shape_.color() : color_per_instance_[info.instance_handle];
183+
std::cout << instance_name << " Shape has been disposed" << std::endl;
179184
if (std::find(disposed_instances_.begin(), disposed_instances_.end(),
180185
info.instance_handle) == disposed_instances_.end())
181186
{
@@ -184,8 +189,13 @@ void SubscriberApp::on_data_available(
184189
}
185190
else if (info.instance_state == NOT_ALIVE_NO_WRITERS_INSTANCE_STATE)
186191
{
187-
std::cout << shape_.color() << " Shape has been disposed by all publishers" << std::endl;
192+
std::string instance_name = shape_.color() !=
193+
"" ? shape_.color() : color_per_instance_[info.instance_handle];
194+
std::cout << instance_name << " Shape has been disposed by all publishers" << std::endl;
188195
}
196+
197+
// Cleaning shape to avoid reusing old color in logs
198+
shape_.color().clear();
189199
}
190200
}
191201

‎examples/cpp/topic_instances/SubscriberApp.hpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,8 @@ class SubscriberApp : public Application, public DataReaderListener
101101
std::map<InstanceHandle_t, uint32_t> samples_per_instance_;
102102

103103
std::vector<InstanceHandle_t> disposed_instances_;
104+
105+
std::map<InstanceHandle_t, std::string> color_per_instance_;
104106
};
105107

106108
} // namespace topic_instances

‎src/cpp/fastdds/publisher/DataWriterImpl.cpp‎

Lines changed: 23 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1021,16 +1021,32 @@ ReturnCode_t DataWriterImpl::perform_create_new_change(
10211021
{
10221022
uint32_t payload_size = fixed_payload_size_ ? fixed_payload_size_ : type_->calculate_serialized_size(
10231023
data, data_representation_);
1024-
if (!get_free_payload_from_pool(payload_size, payload))
1024+
// Initialize payload to null state
1025+
payload.length = 0;
1026+
payload.max_size = 0;
1027+
payload.data = nullptr;
1028+
payload.payload_owner = nullptr;
1029+
bool should_serialize = (change_kind == ALIVE);
1030+
if (should_serialize)
10251031
{
1026-
return RETCODE_OUT_OF_RESOURCES;
1027-
}
1032+
// Request payload from pool and proceed with serialization
1033+
if (!get_free_payload_from_pool(payload_size, payload))
1034+
{
1035+
// ALIVE changes need a payload and serialization
1036+
return RETCODE_OUT_OF_RESOURCES;
1037+
}
10281038

1029-
if ((ALIVE == change_kind) && !type_->serialize(data, payload, data_representation_))
1039+
if (!type_->serialize(data, payload, data_representation_))
1040+
{
1041+
EPROSIMA_LOG_WARNING(DATA_WRITER, "Data serialization returned false");
1042+
payload_pool_->release_payload(payload);
1043+
return RETCODE_ERROR;
1044+
}
1045+
}
1046+
else
10301047
{
1031-
EPROSIMA_LOG_WARNING(DATA_WRITER, "Data serialization returned false");
1032-
payload_pool_->release_payload(payload);
1033-
return RETCODE_ERROR;
1048+
// If not serializable (UNREGISTER or DISPOSE), the handle must be defined
1049+
assert(handle.isDefined());
10341050
}
10351051
}
10361052

‎src/cpp/rtps/reader/StatefulReader.cpp‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -649,6 +649,16 @@ bool StatefulReader::process_data_msg(
649649
}
650650
datasharing_pool->get_datasharing_change(change->serializedPayload, *change_to_add);
651651
}
652+
else if (change->serializedPayload.length == 0 && change->kind != ChangeKind_t::ALIVE &&
653+
change->instanceHandle.isDefined())
654+
{
655+
// A UNREGISTER or DISPOSE status change was sent without payload, but calling get_payload with size 0 might fail
656+
// depending on the configured payload pool. However, those operations are still valid if and only if instanceHandle
657+
// is defined so they are handled in this special case
658+
// These conditions were already checked in change_is_relevant_for_filter, but it makes sense to have a proper case
659+
// here
660+
change_to_add->serializedPayload.length = 0;
661+
}
652662
else if (payload_pool_->get_payload(change->serializedPayload, change_to_add->serializedPayload))
653663
{
654664
if (change->serializedPayload.payload_owner == nullptr)
@@ -671,7 +681,10 @@ bool StatefulReader::process_data_msg(
671681
{
672682
EPROSIMA_LOG_INFO(RTPS_MSG_IN,
673683
IDSTRING "Change " << change_to_add->sequenceNumber << " not added to history");
674-
change_to_add->serializedPayload.payload_owner->release_payload(change_to_add->serializedPayload);
684+
if (change_to_add->serializedPayload.payload_owner)
685+
{
686+
change_to_add->serializedPayload.payload_owner->release_payload(change_to_add->serializedPayload);
687+
}
675688
change_pool_->release_cache(change_to_add);
676689
return false;
677690
}

‎src/cpp/rtps/reader/StatelessReader.cpp‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -658,6 +658,16 @@ bool StatelessReader::process_data_msg(
658658
}
659659
datasharing_pool->get_datasharing_change(change->serializedPayload, *change_to_add);
660660
}
661+
else if (change->serializedPayload.length == 0 && change->kind != ChangeKind_t::ALIVE &&
662+
change->instanceHandle.isDefined())
663+
{
664+
// A UNREGISTER or DISPOSE status change was sent without payload, but calling get_payload with size 0 might fail
665+
// depending on the configured payload pool. However, those operations are still valid iff instanceHandle is defined
666+
// so they are handled in this special case
667+
// These conditions were already checked in change_is_relevant_for_filter, but it makes sense to have a proper case
668+
// here
669+
change_to_add->serializedPayload.length = 0;
670+
}
661671
else if (payload_pool_->get_payload(change->serializedPayload, change_to_add->serializedPayload))
662672
{
663673
if (change->serializedPayload.payload_owner == nullptr)
@@ -680,7 +690,10 @@ bool StatelessReader::process_data_msg(
680690
{
681691
EPROSIMA_LOG_INFO(RTPS_MSG_IN,
682692
IDSTRING "MessageReceiver not add change " << change_to_add->sequenceNumber);
683-
change_to_add->serializedPayload.payload_owner->release_payload(change_to_add->serializedPayload);
693+
if (change_to_add->serializedPayload.payload_owner)
694+
{
695+
change_to_add->serializedPayload.payload_owner->release_payload(change_to_add->serializedPayload);
696+
}
684697
change_pool_->release_cache(change_to_add);
685698
return false;
686699
}

0 commit comments

Comments
 (0)