Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[20401] Check History QoS inconsistencies (backport #4375) #4408

Merged
merged 6 commits into from
Apr 10, 2024
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ bool HelloWorldPublisher::init()
DataWriterQos wqos;
wqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
wqos.history().kind = KEEP_LAST_HISTORY_QOS;
wqos.history().depth = 30;
wqos.history().depth = 20;
wqos.resource_limits().max_samples = 50;
wqos.resource_limits().max_samples_per_instance = 20;
wqos.reliable_writer_qos().times.heartbeatPeriod.seconds = 2;
Expand Down
15 changes: 15 additions & 0 deletions src/cpp/fastdds/publisher/DataWriterImpl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1704,6 +1704,21 @@ ReturnCode_t DataWriterImpl::check_qos(
logError(RTPS_QOS_CHECK, "DATA_SHARING cannot be used with memory policies other than PREALLOCATED.");
return ReturnCode_t::RETCODE_INCONSISTENT_POLICY;
}
if (qos.history().kind == KEEP_LAST_HISTORY_QOS && qos.history().depth <= 0)
{
logError(RTPS_QOS_CHECK, "HISTORY DEPTH must be higher than 0 if HISTORY KIND is KEEP_LAST.");
return ReturnCode_t::RETCODE_INCONSISTENT_POLICY;
}
if (qos.history().kind == KEEP_LAST_HISTORY_QOS && qos.history().depth > 0 &&
qos.resource_limits().max_samples_per_instance > 0 &&
qos.history().depth > qos.resource_limits().max_samples_per_instance)
{
logWarning(RTPS_QOS_CHECK,
"HISTORY DEPTH '" << qos.history().depth <<
"' is inconsistent with max_samples_per_instance: '" << qos.resource_limits().max_samples_per_instance <<
"'. Consistency rule: depth <= max_samples_per_instance." <<
" Effectively using max_samples_per_instance as depth.");
}
return ReturnCode_t::RETCODE_OK;
}

Expand Down
15 changes: 15 additions & 0 deletions src/cpp/fastdds/subscriber/DataReaderImpl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1321,6 +1321,21 @@ ReturnCode_t DataReaderImpl::check_qos (
logError(DDS_QOS_CHECK, "unique_network_request cannot be set along specific locators");
return ReturnCode_t::RETCODE_INCONSISTENT_POLICY;
}
if (qos.history().kind == KEEP_LAST_HISTORY_QOS && qos.history().depth <= 0)
{
logError(RTPS_QOS_CHECK, "HISTORY DEPTH must be higher than 0 if HISTORY KIND is KEEP_LAST.");
return ReturnCode_t::RETCODE_INCONSISTENT_POLICY;
}
if (qos.history().kind == KEEP_LAST_HISTORY_QOS && qos.history().depth > 0 &&
qos.resource_limits().max_samples_per_instance > 0 &&
qos.history().depth > qos.resource_limits().max_samples_per_instance)
{
logWarning(RTPS_QOS_CHECK,
"HISTORY DEPTH '" << qos.history().depth <<
"' is inconsistent with max_samples_per_instance: '" << qos.resource_limits().max_samples_per_instance <<
"'. Consistency rule: depth <= max_samples_per_instance." <<
" Effectively using max_samples_per_instance as depth.");
}
return ReturnCode_t::RETCODE_OK;
}

Expand Down
11 changes: 11 additions & 0 deletions test/blackbox/common/DDSBlackboxTestsDataReader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,17 @@ TEST(DDSDataReader, ConsistentReliabilityWhenIntraprocess)
xmlparser::XMLProfileManager::library_settings(library_settings);
}

/**
* This is a regression test for issue https://eprosima.easyredmine.com/issues/20504.
* It checks that a DataReader be created with default Qos and a large history depth.
*/
TEST(DDSDataReader, default_qos_large_history_depth)
{
PubSubReader<HelloWorldPubSubType> reader(TEST_TOPIC_NAME);
reader.history_depth(1000).init();
ASSERT_TRUE(reader.isInitialized());
}

#ifdef INSTANTIATE_TEST_SUITE_P
#define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
#else
Expand Down
30 changes: 25 additions & 5 deletions test/blackbox/common/DDSBlackboxTestsDataWriter.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -298,6 +298,7 @@ TEST(DDSDataWriter, OfferedDeadlineMissedListener)
* - Only affects TRANSPORT case (UDP or SHM communication, data_sharing and intraprocess disabled)
* - Destruction order matters: writer must be destroyed before reader (otherwise heartbeats would no be sent while
* destroying the writer)
* Edit: this test has been updated to ensure that HistoryQoS and ResourceLimitQoS constraints are met (#20401).
*/
TEST(DDSDataWriter, HeartbeatWhileDestruction)
{
Expand All @@ -310,13 +311,21 @@ TEST(DDSDataWriter, HeartbeatWhileDestruction)
// A high number of samples increases the probability of the data race to occur
size_t n_samples = 1000;

reader.reliability(RELIABLE_RELIABILITY_QOS).durability_kind(TRANSIENT_LOCAL_DURABILITY_QOS).init();
reader.reliability(RELIABLE_RELIABILITY_QOS)
.durability_kind(TRANSIENT_LOCAL_DURABILITY_QOS)
.init();
ASSERT_TRUE(reader.isInitialized());

writer.reliability(RELIABLE_RELIABILITY_QOS).durability_kind(TRANSIENT_LOCAL_DURABILITY_QOS).history_kind(
KEEP_LAST_HISTORY_QOS).history_depth(static_cast<int32_t>(n_samples)).heartbeat_period_seconds(0).
heartbeat_period_nanosec(
20 * 1000).init();
writer.reliability(RELIABLE_RELIABILITY_QOS)
.durability_kind(TRANSIENT_LOCAL_DURABILITY_QOS)
.history_kind(KEEP_LAST_HISTORY_QOS)
.history_depth(static_cast<int32_t>(n_samples))
.resource_limits_max_samples(static_cast<int32_t>(n_samples))
.resource_limits_max_instances(static_cast<int32_t>(1))
.resource_limits_max_samples_per_instance(static_cast<int32_t>(n_samples))
.heartbeat_period_seconds(0)
.heartbeat_period_nanosec(20 * 1000)
.init();
ASSERT_TRUE(writer.isInitialized());

reader.wait_discovery();
Expand All @@ -330,6 +339,17 @@ TEST(DDSDataWriter, HeartbeatWhileDestruction)
}
}

/**
* This is a regression test for issue https://eprosima.easyredmine.com/issues/20504.
* It checks that a DataWriter be created with default Qos and a large history depth.
*/
TEST(DDSDataWriter, default_qos_large_history_depth)
{
PubSubWriter<HelloWorldPubSubType> writer(TEST_TOPIC_NAME);
writer.history_depth(1000).init();
ASSERT_TRUE(writer.isInitialized());
}

#ifdef INSTANTIATE_TEST_SUITE_P
#define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
#else
Expand Down
11 changes: 6 additions & 5 deletions test/unittest/dds/profiles/test_xml_profiles.xml
Original file line number Diff line number Diff line change
Expand Up @@ -261,7 +261,7 @@
<name>default_name</name>
</rtps>
</participant>

<publisher profile_name="test_publisher_profile">
<topic>
<kind>NO_KEY</kind>
Expand Down Expand Up @@ -481,7 +481,8 @@
<increment>0</increment>
</matchedSubscribersAllocation>
</publisher>


<!-- This profile has been updated to ensure that HistoryQoS and ResourceLimitQoS constraints are met (#20401) -->
<subscriber profile_name="test_subscriber_profile">
<topic>
<kind>NO_KEY</kind>
Expand All @@ -492,9 +493,9 @@
<depth>500</depth>
</historyQos>
<resourceLimitsQos>
<max_samples>10</max_samples>
<max_samples>2500</max_samples>
<max_instances>5</max_instances>
<max_samples_per_instance>2</max_samples_per_instance>
<max_samples_per_instance>500</max_samples_per_instance>
<allocated_samples>10</allocated_samples>
</resourceLimitsQos>
</topic>
Expand Down Expand Up @@ -578,7 +579,7 @@
<increment>5</increment>
</matchedPublishersAllocation>
</subscriber>

<subscriber profile_name="test_default_subscriber_profile" is_default_profile="true">
<topic>
<kind>WITH_KEY</kind>
Expand Down
85 changes: 84 additions & 1 deletion test/unittest/dds/publisher/DataWriterTests.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.

#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <memory>
#include <mutex>
#include <thread>

Expand All @@ -24,6 +27,7 @@
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipantListener.hpp>
#include <fastdds/dds/domain/qos/DomainParticipantQos.hpp>
#include <fastdds/dds/log/Log.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
Expand All @@ -36,7 +40,6 @@
#include <fastdds/rtps/writer/StatefulWriter.h>

#include <fastdds/publisher/DataWriterImpl.hpp>

#include "../../logging/mock/MockConsumer.h"

namespace eprosima {
Expand Down Expand Up @@ -665,6 +668,17 @@ TEST(DataWriterTests, InvalidQos)
qos.endpoint().history_memory_policy = eprosima::fastrtps::rtps::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
EXPECT_EQ(ReturnCode_t::RETCODE_OK, datawriter->set_qos(qos));

qos = DATAWRITER_QOS_DEFAULT;
qos.history().kind = KEEP_LAST_HISTORY_QOS;
qos.history().depth = 0;
EXPECT_EQ(inconsistent_code, datawriter->set_qos(qos)); // KEEP LAST 0 is inconsistent
qos.history().depth = 2;
EXPECT_EQ(ReturnCode_t::RETCODE_OK, datawriter->set_qos(qos)); // KEEP LAST 2 is OK
// KEEP LAST 2000 but max_samples_per_instance default (400) is inconsistent but right now it only shows a warning
// This test will fail whenever we enforce the consistency between depth and max_samples_per_instance.
qos.history().depth = 2000;
EXPECT_EQ(ReturnCode_t::RETCODE_OK, datawriter->set_qos(qos));

ASSERT_TRUE(publisher->delete_datawriter(datawriter) == ReturnCode_t::RETCODE_OK);
ASSERT_TRUE(participant->delete_topic(topic) == ReturnCode_t::RETCODE_OK);
ASSERT_TRUE(participant->delete_publisher(publisher) == ReturnCode_t::RETCODE_OK);
Expand Down Expand Up @@ -1509,6 +1523,75 @@ TEST_F(DataWriterUnsupportedTests, UnsupportedDataWriterMethods)
ASSERT_EQ(DomainParticipantFactory::get_instance()->delete_participant(participant), ReturnCode_t::RETCODE_OK);
}

TEST(DataWriterTests, history_depth_max_samples_per_instance_warning)
{

/* Setup log so it may catch the expected warning */
Log::ClearConsumers();
MockConsumer* mockConsumer = new MockConsumer("RTPS_QOS_CHECK");
Log::RegisterConsumer(std::unique_ptr<LogConsumer>(mockConsumer));
Log::SetVerbosity(Log::Warning);

/* Create a participant, topic, and a publisher */
DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0,
PARTICIPANT_QOS_DEFAULT);
ASSERT_NE(participant, nullptr);

TypeSupport type(new TopicDataTypeMock());
type.register_type(participant);

Topic* topic = participant->create_topic("footopic", type.get_type_name(), TOPIC_QOS_DEFAULT);
ASSERT_NE(topic, nullptr);

Publisher* publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
ASSERT_NE(publisher, nullptr);

/* Create a datawriter with the QoS that should generate a warning */
DataWriterQos qos;
qos.history().depth = 10;
qos.resource_limits().max_samples_per_instance = 5;
DataWriter* datawriter_1 = publisher->create_datawriter(topic, qos);
ASSERT_NE(datawriter_1, nullptr);

/* Check that the config generated a warning */
auto wait_for_log_entries =
[&mockConsumer](const uint32_t amount, const uint32_t retries, const uint32_t wait_ms) -> size_t
{
size_t entries = 0;
for (uint32_t i = 0; i < retries; i++)
{
entries = mockConsumer->ConsumedEntries().size();
if (entries >= amount)
{
break;
}
std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
}
return entries;
};

const size_t expected_entries = 1;
const uint32_t retries = 4;
const uint32_t wait_ms = 25;
ASSERT_EQ(wait_for_log_entries(expected_entries, retries, wait_ms), expected_entries);

/* Check that the datawriter can send data */
FooType data;
ASSERT_EQ(ReturnCode_t::RETCODE_OK, datawriter_1->write(&data, HANDLE_NIL));

/* Check that a correctly initialized writer does not produce any warning */
qos.history().depth = 10;
qos.resource_limits().max_samples_per_instance = 10;
DataWriter* datawriter_2 = publisher->create_datawriter(topic, qos);
ASSERT_NE(datawriter_2, nullptr);
ASSERT_EQ(wait_for_log_entries(expected_entries, retries, wait_ms), expected_entries);

/* Tear down */
participant->delete_contained_entities();
DomainParticipantFactory::get_instance()->delete_participant(participant);
Log::KillThread();
}

} // namespace dds
} // namespace fastdds
} // namespace eprosima
Expand Down
Loading
Loading