Skip to content

Commit ea365df

Browse files
raulojeda22mergify[bot]
raulojeda22
authored and
mergify[bot]
committed
Fix DataReader history enforcement to respect max_samples_per_instance (#6228)
* Fix DataReader history enforcement to respect max_samples_per_instance Signed-off-by: Raül <raulojeda@eprosima.com> * Take into account LENGTH_UNLIMITED, go back to max_samples allocation and add tests accordingly Signed-off-by: Raül <raulojeda@eprosima.com> * Uncrustify Signed-off-by: Raül <raulojeda@eprosima.com> * More uncrustify Signed-off-by: Raül <raulojeda@eprosima.com> * Refactor history depth vs mspi tests Signed-off-by: Raül <raulojeda@eprosima.com> * Uncrustify Signed-off-by: Raül <raulojeda@eprosima.com> * Update src/cpp/fastdds/publisher/DataWriterHistory.cpp Co-authored-by: Miguel Company <miguelcompany@eprosima.com> Signed-off-by: Raül Ojeda Gandia <raulojeda@eprosima.com> * Refactor tests and add keyed keep_all tests Signed-off-by: Raül <raulojeda@eprosima.com> * Update src/cpp/fastdds/publisher/DataWriterHistory.cpp Co-authored-by: Miguel Company <miguelcompany@eprosima.com> Signed-off-by: Raül Ojeda Gandia <raulojeda@eprosima.com> * New uncrustify Signed-off-by: Raül <raulojeda@eprosima.com> --------- Signed-off-by: Raül <raulojeda@eprosima.com> Signed-off-by: Raül Ojeda Gandia <raulojeda@eprosima.com> Co-authored-by: Miguel Company <miguelcompany@eprosima.com> (cherry picked from commit 30b6351) # Conflicts: # test/unittest/dds/subscriber/DataReaderTests.cpp
1 parent 22a656d commit ea365df

File tree

7 files changed

+409
-28
lines changed

7 files changed

+409
-28
lines changed

‎src/cpp/fastdds/publisher/DataWriterHistory.cpp‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
#include <fastdds/rtps/common/Time_t.hpp>
2828
#include <fastdds/rtps/writer/RTPSWriter.hpp>
2929

30+
#include <rtps/history/HistoryAttributesExtension.hpp>
3031
#include <rtps/writer/BaseWriter.hpp>
3132

3233
namespace eprosima {
@@ -48,13 +49,23 @@ HistoryAttributes DataWriterHistory::to_history_attributes(
4849

4950
if (history_qos.kind != KEEP_ALL_HISTORY_QOS)
5051
{
51-
max_samples = history_qos.depth;
52+
max_samples = get_min_max_samples(history_qos.depth, resource_limits_qos.max_samples_per_instance);
5253
if (topic_kind != NO_KEY)
5354
{
54-
max_samples *= resource_limits_qos.max_instances;
55+
if (0 < resource_limits_qos.max_instances)
56+
{
57+
max_samples *= resource_limits_qos.max_instances;
58+
}
59+
else
60+
{
61+
max_samples = LENGTH_UNLIMITED;
62+
}
5563
}
5664

57-
initial_samples = std::min(initial_samples, max_samples);
65+
if (0 < initial_samples)
66+
{
67+
initial_samples = get_min_max_samples(initial_samples, max_samples);
68+
}
5869
}
5970

6071
return HistoryAttributes(mempolicy, payloadMaxSize, initial_samples, max_samples, extra_samples);

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

Lines changed: 23 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -336,6 +336,20 @@ ReturnCode_t DataWriterImpl::enable()
336336
datasharing.add_domain_id(utils::default_domain_id());
337337
}
338338
w_att.endpoint.set_data_sharing_configuration(datasharing);
339+
340+
// Update pool config for KEEP_ALL when max_samples is infinite
341+
if ((0 == pool_config_.maximum_size) && (KEEP_ALL_HISTORY_QOS == qos_.history().kind))
342+
{
343+
// Override infinite with old default value for max_samples + extra samples
344+
pool_config_.maximum_size = 5000;
345+
if (0 < qos_.resource_limits().extra_samples)
346+
{
347+
pool_config_.maximum_size += static_cast<uint32_t>(qos_.resource_limits().extra_samples);
348+
}
349+
EPROSIMA_LOG_ERROR(DATA_WRITER,
350+
"DataWriter with KEEP_ALL history and infinite max_samples is not compatible with DataSharing. "
351+
"Setting max_samples to " << pool_config_.maximum_size);
352+
}
339353
}
340354
else
341355
{
@@ -2043,7 +2057,8 @@ ReturnCode_t DataWriterImpl::check_qos(
20432057
{
20442058
EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK,
20452059
"HISTORY DEPTH '" << qos.history().depth <<
2046-
"' is inconsistent with max_samples_per_instance: '" << qos.resource_limits().max_samples_per_instance <<
2060+
"' is inconsistent with max_samples_per_instance: '" <<
2061+
qos.resource_limits().max_samples_per_instance <<
20472062
"'. Consistency rule: depth <= max_samples_per_instance." <<
20482063
" Effectively using max_samples_per_instance as depth.");
20492064
}
@@ -2053,19 +2068,19 @@ ReturnCode_t DataWriterImpl::check_qos(
20532068
ReturnCode_t DataWriterImpl::check_allocation_consistency(
20542069
const DataWriterQos& qos)
20552070
{
2056-
if ((qos.resource_limits().max_samples > 0) &&
2057-
(qos.resource_limits().max_samples <
2058-
(qos.resource_limits().max_instances * qos.resource_limits().max_samples_per_instance)))
2071+
if ((qos.resource_limits().max_instances <= 0 || qos.resource_limits().max_samples_per_instance <= 0) &&
2072+
(qos.resource_limits().max_samples > 0))
20592073
{
20602074
EPROSIMA_LOG_ERROR(DDS_QOS_CHECK,
2061-
"max_samples should be greater than max_instances * max_samples_per_instance");
2075+
"max_samples should be infinite when max_instances or max_samples_per_instance are infinite");
20622076
return RETCODE_INCONSISTENT_POLICY;
20632077
}
2064-
if ((qos.resource_limits().max_instances <= 0 || qos.resource_limits().max_samples_per_instance <= 0) &&
2065-
(qos.resource_limits().max_samples > 0))
2078+
if ((qos.resource_limits().max_samples > 0) &&
2079+
(qos.resource_limits().max_samples <
2080+
(qos.resource_limits().max_instances * qos.resource_limits().max_samples_per_instance)))
20662081
{
20672082
EPROSIMA_LOG_ERROR(DDS_QOS_CHECK,
2068-
"max_samples should be infinite when max_instances or max_samples_per_instance are infinite");
2083+
"max_samples should be greater than max_instances * max_samples_per_instance");
20692084
return RETCODE_INCONSISTENT_POLICY;
20702085
}
20712086
return RETCODE_OK;

‎src/cpp/fastdds/subscriber/DataReaderImpl.cpp‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1616,7 +1616,8 @@ ReturnCode_t DataReaderImpl::check_qos(
16161616
{
16171617
EPROSIMA_LOG_WARNING(RTPS_QOS_CHECK,
16181618
"HISTORY DEPTH '" << qos.history().depth <<
1619-
"' is inconsistent with max_samples_per_instance: '" << qos.resource_limits().max_samples_per_instance <<
1619+
"' is inconsistent with max_samples_per_instance: '" <<
1620+
qos.resource_limits().max_samples_per_instance <<
16201621
"'. Consistency rule: depth <= max_samples_per_instance." <<
16211622
" Effectively using max_samples_per_instance as depth.");
16221623
}
@@ -1626,19 +1627,19 @@ ReturnCode_t DataReaderImpl::check_qos(
16261627
ReturnCode_t DataReaderImpl::check_allocation_consistency(
16271628
const DataReaderQos& qos)
16281629
{
1629-
if ((qos.resource_limits().max_samples > 0) &&
1630-
(qos.resource_limits().max_samples <
1631-
(qos.resource_limits().max_instances * qos.resource_limits().max_samples_per_instance)))
1630+
if ((qos.resource_limits().max_instances <= 0 || qos.resource_limits().max_samples_per_instance <= 0) &&
1631+
(qos.resource_limits().max_samples > 0))
16321632
{
16331633
EPROSIMA_LOG_ERROR(DDS_QOS_CHECK,
1634-
"max_samples should be greater than max_instances * max_samples_per_instance");
1634+
"max_samples should be infinite when max_instances or max_samples_per_instance are infinite");
16351635
return RETCODE_INCONSISTENT_POLICY;
16361636
}
1637-
if ((qos.resource_limits().max_instances <= 0 || qos.resource_limits().max_samples_per_instance <= 0) &&
1638-
(qos.resource_limits().max_samples > 0))
1637+
if ((qos.resource_limits().max_samples > 0) &&
1638+
(qos.resource_limits().max_samples <
1639+
(qos.resource_limits().max_instances * qos.resource_limits().max_samples_per_instance)))
16391640
{
16401641
EPROSIMA_LOG_ERROR(DDS_QOS_CHECK,
1641-
"max_samples should be infinite when max_instances or max_samples_per_instance are infinite");
1642+
"max_samples should be greater than max_instances * max_samples_per_instance");
16421643
return RETCODE_INCONSISTENT_POLICY;
16431644
}
16441645
return RETCODE_OK;

‎src/cpp/fastdds/subscriber/history/DataReaderHistory.cpp‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -96,9 +96,16 @@ DataReaderHistory::DataReaderHistory(
9696
{
9797
resource_limited_qos_.max_instances = 1;
9898
resource_limited_qos_.max_samples_per_instance = resource_limited_qos_.max_samples;
99-
key_changes_allocation_.initial = resource_limited_qos_.allocated_samples;
100-
key_changes_allocation_.maximum = resource_limited_qos_.max_samples;
10199

100+
if (0 < resource_limited_qos_.allocated_samples)
101+
{
102+
key_changes_allocation_.initial = resource_limited_qos_.allocated_samples;
103+
}
104+
105+
if (resource_limited_qos_.max_samples_per_instance < std::numeric_limits<int32_t>::max())
106+
{
107+
key_changes_allocation_.maximum = resource_limited_qos_.max_samples_per_instance;
108+
}
102109
instances_.emplace(c_InstanceHandle_Unknown,
103110
std::make_shared<DataReaderInstance>(key_changes_allocation_, key_writers_allocation_));
104111
data_available_instances_[c_InstanceHandle_Unknown] = instances_[c_InstanceHandle_Unknown];
@@ -260,7 +267,8 @@ bool DataReaderHistory::received_change_keep_last(
260267
if (find_key(a_change->instanceHandle, vit))
261268
{
262269
DataReaderInstance::ChangeCollection& instance_changes = vit->second->cache_changes;
263-
if (instance_changes.size() < static_cast<size_t>(history_qos_.depth))
270+
auto effective_depth = std::min(history_qos_.depth, resource_limited_qos_.max_samples_per_instance);
271+
if (instance_changes.size() < static_cast<size_t>(effective_depth))
264272
{
265273
ret_value = true;
266274
}
@@ -795,7 +803,8 @@ bool DataReaderHistory::completed_change_keep_last(
795803
{
796804
bool ret_value = false;
797805
DataReaderInstance::ChangeCollection& instance_changes = instance.cache_changes;
798-
if (instance_changes.size() < static_cast<size_t>(history_qos_.depth))
806+
auto effective_depth = std::min(history_qos_.depth, resource_limited_qos_.max_samples_per_instance);
807+
if (instance_changes.size() < static_cast<size_t>(effective_depth))
799808
{
800809
ret_value = true;
801810
}

‎src/cpp/rtps/history/HistoryAttributesExtension.hpp‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,21 @@ static inline ResourceLimitedContainerConfig resource_limits_from_history(
4949
};
5050
}
5151

52+
/**
53+
* Get the minimum value between two sample counts, considering that <= 0 means unlimited.
54+
*
55+
* @param a First sample count.
56+
* @param b Second sample count.
57+
*
58+
* @return Minimum sample count.
59+
*/
60+
static constexpr int32_t get_min_max_samples(
61+
int32_t a,
62+
int32_t b)
63+
{
64+
return (a > 0 && b > 0) ? (a < b ? a : b) : (a > 0 ? a : b);
65+
}
66+
5267
} // namespace rtps
5368
} // namespace fastdds
5469
} // namespace eprosima

‎test/mock/dds/DataWriterHistory/fastdds/publisher/DataWriterHistory.hpp‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,8 @@
3434

3535
#include <fastdds/publisher/history/DataWriterInstance.hpp>
3636

37+
#include <rtps/history/HistoryAttributesExtension.hpp>
38+
3739
namespace eprosima {
3840
namespace fastdds {
3941
namespace dds {
@@ -57,13 +59,20 @@ class DataWriterHistory : public WriterHistory
5759

5860
if (history_qos.kind != KEEP_ALL_HISTORY_QOS)
5961
{
60-
max_samples = history_qos.depth;
62+
max_samples = get_min_max_samples(history_qos.depth, resource_limits_qos.max_samples_per_instance);
6163
if (topic_kind != NO_KEY)
6264
{
63-
max_samples *= resource_limits_qos.max_instances;
65+
if (0 < resource_limits_qos.max_instances)
66+
{
67+
max_samples *= resource_limits_qos.max_instances;
68+
}
69+
else
70+
{
71+
max_samples = LENGTH_UNLIMITED;
72+
}
6473
}
6574

66-
initial_samples = std::min(initial_samples, max_samples);
75+
initial_samples = get_min_max_samples(initial_samples, max_samples);
6776
}
6877

6978
return HistoryAttributes(mempolicy, payloadMaxSize, initial_samples, max_samples, extra_samples);

0 commit comments

Comments
 (0)