Fix publishing on physical data topic (#2581)

* Refs #14124: Create statistics datawriter after enabling the DomainParticipantImpl

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Check whether PDP is enabled before stopping announcement

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Check enable return before creating statistics data writers

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Add regression test for removing not enabled RTPSParticipant

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Add test for deleting not enabled DDS entities

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Apply suggestions

Signed-off-by: Eduardo Ponz <[email protected]>

* Refs #14124: Apply suggestions

Signed-off-by: Eduardo Ponz <[email protected]>

Author: EduPonz

include/fastdds/rtps/builtin/discovery/participant/PDP.h

@@ -21,6 +21,7 @@
 #define _FASTDDS_RTPS_PDP_H_
 #ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
 
+#include <atomic>
 #include <mutex>
 #include <functional>
 
@@ -97,6 +98,11 @@ class PDP
     bool initPDP(
             RTPSParticipantImpl* part);
 
+    /**
+     * @brief Enable the Participant Discovery Protocol
+     *
+     * @return true if enabled correctly, or if already enabled; false otherwise
+     */
     bool enable();
 
     virtual bool init(
@@ -402,7 +408,7 @@ class PDP
     //!To protect callbacks (ParticipantProxyData&)
     std::mutex callback_mtx_;
     //!Tell if object is enabled
-    bool enable_ = false;
+    std::atomic<bool> enabled_ {false};
 
     /**
      * Adds an entry to the collection of participant proxy information.

src/cpp/rtps/builtin/discovery/participant/PDP.cpp

@@ -408,6 +408,12 @@ bool PDP::initPDP(
 
 bool PDP::enable()
 {
+    // It is safe to call enable() on already enable PDPs
+    if (enabled_)
+    {
+        return true;
+    }
+
     // Create lease events on already created proxy data objects
     for (ParticipantProxyData* pool_item : participant_proxies_pool_)
     {
@@ -430,7 +436,7 @@ bool PDP::enable()
 
     set_initial_announcement_interval();
 
-    enable_  = true;
+    enabled_.store(true);
     // Notify "self-discovery"
     getRTPSParticipant()->on_entity_discovery(mp_RTPSParticipant->getGuid(),
             get_participant_proxy_data(mp_RTPSParticipant->getGuid().guidPrefix)->m_properties);
@@ -443,7 +449,7 @@ void PDP::announceParticipantState(
         bool dispose,
         WriteParams& wparams)
 {
-    if (enable_)
+    if (enabled_)
     {
         // logInfo(RTPS_PDP, "Announcing RTPSParticipant State (new change: " << new_change << ")");
         CacheChange_t* change = nullptr;
@@ -543,12 +549,18 @@ void PDP::announceParticipantState(
 
 void PDP::stopParticipantAnnouncement()
 {
-    resend_participant_info_event_->cancel_timer();
+    if (resend_participant_info_event_)
+    {
+        resend_participant_info_event_->cancel_timer();
+    }
 }
 
 void PDP::resetParticipantAnnouncement()
 {
-    resend_participant_info_event_->restart_timer();
+    if (resend_participant_info_event_)
+    {
+        resend_participant_info_event_->restart_timer();
+    }
 }
 
 bool PDP::has_reader_proxy_data(

src/cpp/rtps/builtin/discovery/participant/PDPClient.cpp

@@ -410,7 +410,7 @@ void PDPClient::announceParticipantState(
         bool dispose,
         WriteParams& )
 {
-    if (enable_)
+    if (enabled_)
     {
         /*
            Protect writer sequence number. Make sure in order to prevent AB BA deadlock that the

src/cpp/rtps/builtin/discovery/participant/PDPServer.cpp

@@ -538,7 +538,7 @@ void PDPServer::announceParticipantState(
         bool dispose /* = false */,
         WriteParams& )
 {
-    if (enable_)
+    if (enabled_)
     {
         logInfo(RTPS_PDP_SERVER,
                 "Announcing Server " << mp_RTPSParticipant->getGuid() << " (new change: " << new_change << ")");

src/cpp/rtps/builtin/discovery/participant/PDPSimple.cpp

@@ -214,7 +214,7 @@ void PDPSimple::announceParticipantState(
         bool dispose,
         WriteParams& wp)
 {
-    if (enable_)
+    if (enabled_)
     {
         PDP::announceParticipantState(new_change, dispose, wp);
 

src/cpp/statistics/fastdds/domain/DomainParticipantImpl.cpp

@@ -202,12 +202,12 @@ ReturnCode_t DomainParticipantImpl::disable_statistics_datawriter(
 
 ReturnCode_t DomainParticipantImpl::enable()
 {
-    create_statistics_builtin_entities();
     ReturnCode_t ret = efd::DomainParticipantImpl::enable();
 
     if (ReturnCode_t::RETCODE_OK == ret)
     {
         rtps_participant_->add_statistics_listener(statistics_listener_, participant_statistics_mask);
+        create_statistics_builtin_entities();
     }
 
     return ret;

src/cpp/statistics/rtps/StatisticsBase.cpp

@@ -492,7 +492,7 @@ void StatisticsParticipantImpl::on_entity_discovery(
      *
      */
     auto get_physical_property_value =
-            [](const dds::ParameterPropertyList_t& properties, const std::string& property_name)
+            [](const dds::ParameterPropertyList_t& properties, const std::string& property_name) -> std::string
             {
                 auto property = std::find_if(
                     properties.begin(),

test/blackbox/common/DDSBlackboxTestsBasic.cpp

@@ -0,0 +1,114 @@
+// Copyright 2022 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <gtest/gtest.h>
+
+#include <fastdds/dds/domain/DomainParticipant.hpp>
+#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
+#include <fastdds/dds/domain/qos/DomainParticipantFactoryQos.hpp>
+#include <fastdds/dds/domain/qos/DomainParticipantQos.hpp>
+#include <fastdds/dds/publisher/DataWriter.hpp>
+#include <fastdds/dds/publisher/Publisher.hpp>
+#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
+#include <fastdds/dds/publisher/qos/PublisherQos.hpp>
+#include <fastdds/dds/subscriber/DataReader.hpp>
+#include <fastdds/dds/subscriber/qos/DataReaderQos.hpp>
+#include <fastdds/dds/subscriber/qos/SubscriberQos.hpp>
+#include <fastdds/dds/subscriber/Subscriber.hpp>
+#include <fastdds/dds/topic/qos/TopicQos.hpp>
+#include <fastdds/dds/topic/Topic.hpp>
+#include <fastdds/dds/topic/TypeSupport.hpp>
+#include <fastrtps/types/TypesBase.h>
+
+#include "BlackboxTests.hpp"
+#include "../types/HelloWorldPubSubTypes.h"
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+
+using ReturnCode_t = eprosima::fastrtps::types::ReturnCode_t;
+
+/**
+ * This test checks whether it is safe to delete not enabled DDS entities *
+ */
+TEST(DDSBasic, DeleteDisabledEntities)
+{
+    // Set DomainParticipantFactory to create disabled entities
+    DomainParticipantFactoryQos factory_qos;
+    factory_qos.entity_factory().autoenable_created_entities = false;
+    DomainParticipantFactory* factory = DomainParticipantFactory::get_instance();
+    ASSERT_NE(nullptr, factory);
+    factory->set_qos(factory_qos);
+    DomainParticipantFactoryQos factory_qos_check;
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, factory->get_qos(factory_qos_check));
+    ASSERT_EQ(false, factory_qos_check.entity_factory().autoenable_created_entities);
+
+    // Create a disabled DomainParticipant, setting it to in turn create disable entities
+    DomainParticipantQos participant_qos;
+    participant_qos.entity_factory().autoenable_created_entities = false;
+    DomainParticipant* participant = factory->create_participant((uint32_t)GET_PID() % 230, participant_qos);
+    ASSERT_NE(nullptr, participant);
+    DomainParticipantQos participant_qos_check;
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, participant->get_qos(participant_qos_check));
+    ASSERT_EQ(false, participant_qos_check.entity_factory().autoenable_created_entities);
+
+    // Create a disabled Publisher, setting it to in turn create disable entities
+    PublisherQos publisher_qos;
+    publisher_qos.entity_factory().autoenable_created_entities = false;
+    Publisher* publisher = participant->create_publisher(publisher_qos);
+    ASSERT_NE(nullptr, publisher);
+    PublisherQos publisher_qos_check;
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, publisher->get_qos(publisher_qos_check));
+    ASSERT_EQ(false, publisher_qos_check.entity_factory().autoenable_created_entities);
+
+    // Create a disabled Subscriber, setting it to in turn create disable entities
+    SubscriberQos subscriber_qos;
+    subscriber_qos.entity_factory().autoenable_created_entities = false;
+    Subscriber* subscriber = participant->create_subscriber(subscriber_qos);
+    ASSERT_NE(nullptr, subscriber);
+    SubscriberQos subscriber_qos_check;
+    ASSERT_EQ(ReturnCode_t::RETCODE_OK, subscriber->get_qos(subscriber_qos_check));
+    ASSERT_EQ(false, subscriber_qos_check.entity_factory().autoenable_created_entities);
+
+    // Register type
+    TypeSupport type_support;
+    type_support.reset(new HelloWorldPubSubType());
+    type_support.register_type(participant);
+    ASSERT_NE(nullptr, type_support);
+
+    // Create Topic
+    Topic* topic = participant->create_topic("HelloWorld", type_support.get_type_name(), TOPIC_QOS_DEFAULT);
+    ASSERT_NE(nullptr, topic);
+
+    // Create a disabled DataWriter
+    DataWriter* datawriter = publisher->create_datawriter(topic, DATAWRITER_QOS_DEFAULT);
+    ASSERT_NE(nullptr, datawriter);
+
+    // Create a disabled DataReader
+    DataReader* datareader = subscriber->create_datareader(topic, DATAREADER_QOS_DEFAULT);
+    ASSERT_NE(nullptr, datareader);
+
+    // Delete all entities
+    publisher->delete_datawriter(datawriter);
+    subscriber->delete_datareader(datareader);
+    participant->delete_publisher(publisher);
+    participant->delete_subscriber(subscriber);
+    participant->delete_topic(topic);
+    factory->delete_participant(participant);
+}
+
+} // namespace dds
+} // namespace fastdds
+} // namespace eprosima

test/blackbox/common/RTPSBlackboxTestsBasic.cpp

@@ -12,19 +12,23 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "BlackboxTests.hpp"
+
 #include <chrono>
 #include <thread>
 
 #include <gtest/gtest.h>
 
-#include "BlackboxTests.hpp"
+#include <fastrtps/rtps/attributes/RTPSParticipantAttributes.h>
+#include <fastrtps/rtps/participant/RTPSParticipant.h>
+#include <fastrtps/rtps/RTPSDomain.h>
+#include <fastrtps/transport/test_UDPv4TransportDescriptor.h>
+#include <fastrtps/xmlparser/XMLProfileManager.h>
 
 #include "RTPSAsSocketReader.hpp"
 #include "RTPSAsSocketWriter.hpp"
 #include "RTPSWithRegistrationReader.hpp"
 #include "RTPSWithRegistrationWriter.hpp"
-#include <fastrtps/xmlparser/XMLProfileManager.h>
-#include <fastrtps/transport/test_UDPv4TransportDescriptor.h>
 #include <rtps/transport/test_UDPv4Transport.h>
 
 using namespace eprosima::fastrtps;
@@ -634,6 +638,31 @@ TEST(RTPS, RTPSNetworkInterfaceChangesAtRunTime)
     writer.destroy();
 }
 
+/**
+ * Regression test for checking that a not enabled RTPSParticipant can be removed
+ *
+ * https://github.com/eProsima/Fast-DDS/pull/2171 introduced this regression since with it
+ * the PDP is not enabled until calling BuiltinProtocols::enable(), which is called within
+ * RTPSParticipant::enable(). However, during RTPSDomain::removeRTPSParticipant(), there is a call
+ * to BuiltinProtocols::stopRTPSParticipantAnnouncement(), which in turn calls
+ * PDP::stopRTPSParticipantAnnouncement(). That function ends up accessing a timed event pointer,
+ * which is only instantiated on PDP::enable(). Since the RTPSParticipant was not enabled,
+ * BuiltinProtocols and in turn PDP are not either, meaning that it is not safe to call
+ * PDP::stopRTPSParticipantAnnouncement() on a not enabled PDP.
+ *
+ * The test checks that the necessary guards are in place so that it is safe to call
+ * RTPSDomain::removeRTPSParticipant() o a not enabled RTPSParticipant.
+ */
+TEST(RTPS, RemoveDisabledParticipant)
+{
+    RTPSParticipantAttributes rtps_attr;
+    RTPSParticipant* rtps_participant = RTPSDomain::createParticipant(
+        (uint32_t)GET_PID() % 230, false, rtps_attr, nullptr);
+
+    ASSERT_NE(nullptr, rtps_participant);
+    ASSERT_TRUE(RTPSDomain::removeRTPSParticipant(rtps_participant));
+}
+
 #ifdef INSTANTIATE_TEST_SUITE_P
 #define GTEST_INSTANTIATE_TEST_MACRO(x, y, z, w) INSTANTIATE_TEST_SUITE_P(x, y, z, w)
 #else