Skip to content

[21670] Fix Data Races on DDS-Pipe #145

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.

Sign up for GitHub

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

Merged
juanlofer-eprosima merged 9 commits into from
Jul 14, 2025
Merged

[21670] Fix Data Races on DDS-Pipe #145

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
2b9813d
Refs #21670: Protect on_data_available_lambda_
cferreiragonz May 27, 2025
b5b8d6b
Refs #21670: Set listener in RederCreation
cferreiragonz Jun 17, 2025
3df7f05
Refs #21670: Separate DDS/RTPS Listeners from Base Classes
cferreiragonz Jun 17, 2025
8ed78cd
Refs #21670: Uncrustify
cferreiragonz Jun 17, 2025
6af6a14
Refs #21670: Apply Review
cferreiragonz Jun 18, 2025
8760503
Refs #21670: Apply Review 2
cferreiragonz Jun 19, 2025
71631b6
Refs #21670: Apply review 3
cferreiragonz Jul 11, 2025
514a700
Refs #21670: Avoid using capital letters in DDS Listener
cferreiragonz Jul 11, 2025
2277ab2
Refs #21670: Fix leak
cferreiragonz Jul 11, 2025
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
89 changes: 71 additions & 18 deletions ddspipe_participants/include/ddspipe_participants/participant/dds/CommonParticipant.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ namespace dds {
* @warning This Participant class does not support RPC so far.
* @todo TODO
*/
class CommonParticipant : public core::IParticipant, public fastdds::dds::DomainParticipantListener
class CommonParticipant : public core::IParticipant
Comment thread
juanlofer-eprosima marked this conversation as resolved.
{
public:

Expand Down Expand Up @@ -112,23 +112,62 @@ class CommonParticipant : public core::IParticipant, public fastdds::dds::Domain
// LISTENER METHODS
/////////////////////////

void on_participant_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::ParticipantDiscoveryStatus reason,
const fastdds::rtps::ParticipantBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

void on_data_reader_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::dds::SubscriptionBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

void on_data_writer_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::dds::PublicationBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;
class DdsListener : public fastdds::dds::DomainParticipantListener
{
public:

DDSPIPE_PARTICIPANTS_DllAPI
explicit DdsListener(
std::shared_ptr<SimpleParticipantConfiguration> conf,
std::shared_ptr<core::DiscoveryDatabase> ddb);

/**
* @brief Override method from \c DomainParticipantListener
*
* This method is only used for debugging purposes.
*/
DDSPIPE_PARTICIPANTS_DllAPI
void on_participant_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::ParticipantDiscoveryStatus reason,
const fastdds::rtps::ParticipantBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c DomainParticipantListener .
*
* This method adds to the database the discovered or modified endpoint.
*/
DDSPIPE_PARTICIPANTS_DllAPI
void on_data_reader_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::dds::SubscriptionBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c DomainParticipantListener .
*
* This method adds to the database the discovered or modified endpoint.
*/
DDSPIPE_PARTICIPANTS_DllAPI
void on_data_writer_discovery(
fastdds::dds::DomainParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::dds::PublicationBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

protected:

//! Shared pointer to the configuration of the participant
const std::shared_ptr<SimpleParticipantConfiguration> configuration_;
//! Shared pointer to the discovery database
const std::shared_ptr<core::DiscoveryDatabase> discovery_database_;

};

//! Unique pointer to the internal DDS Participant Listener
std::unique_ptr<fastdds::dds::DomainParticipantListener> dds_participant_listener_;

protected:

Expand All @@ -142,19 +181,33 @@ class CommonParticipant : public core::IParticipant, public fastdds::dds::Domain
const std::shared_ptr<core::PayloadPool>& payload_pool,
const std::shared_ptr<core::DiscoveryDatabase>& discovery_database);

/////////////////////////
// VIRTUAL METHODS
/////////////////////////

/**
* @brief Virtual method that creates a listener for the internal DDS Participant.
* It should be overridden if a different listener is needed.
*/
DDSPIPE_PARTICIPANTS_DllAPI
Comment thread
juanlofer-eprosima marked this conversation as resolved.
virtual std::unique_ptr<fastdds::dds::DomainParticipantListener> create_listener_();

/////////////////////////
// INTERNAL VIRTUAL METHODS
/////////////////////////

DDSPIPE_PARTICIPANTS_DllAPI
virtual
fastdds::dds::DomainParticipantQos
add_qos_properties_(
fastdds::dds::DomainParticipantQos& qos) const;

DDSPIPE_PARTICIPANTS_DllAPI
virtual
fastdds::dds::DomainParticipantQos
reckon_participant_qos_() const;

DDSPIPE_PARTICIPANTS_DllAPI
virtual
fastdds::dds::DomainParticipant*
create_dds_participant_();
Expand Down
62 changes: 44 additions & 18 deletions ...rticipants/include/ddspipe_participants/participant/dynamic_types/DynTypesParticipant.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,31 +74,57 @@ class DynTypesParticipant : public rtps::SimpleParticipant
std::shared_ptr<core::IReader> create_reader(
const core::ITopic& topic) override;

DDSPIPE_PARTICIPANTS_DllAPI
void on_reader_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::rtps::SubscriptionBuiltinTopicData& info,
bool& should_be_ignored) override;

DDSPIPE_PARTICIPANTS_DllAPI
void on_writer_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::rtps::PublicationBuiltinTopicData& info,
bool& should_be_ignored) override;
class DynTypesRtpsListener : public rtps::CommonParticipant::RtpsListener
{
public:

DDSPIPE_PARTICIPANTS_DllAPI
explicit DynTypesRtpsListener(
std::shared_ptr<ParticipantConfiguration> conf,
std::shared_ptr<core::DiscoveryDatabase> ddb,
std::shared_ptr<InternalReader> internal_reader);

DDSPIPE_PARTICIPANTS_DllAPI
void on_reader_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::rtps::SubscriptionBuiltinTopicData& info,
bool& should_be_ignored) override;

DDSPIPE_PARTICIPANTS_DllAPI
void on_writer_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::rtps::PublicationBuiltinTopicData& info,
bool& should_be_ignored) override;

//! Type Object Reader getter
inline std::shared_ptr<InternalReader> type_object_reader() const
{
return type_object_reader_;
}

protected:

//! Copy of Type Object Internal Reader
std::shared_ptr<InternalReader> type_object_reader_;
//! Received types set
std::set<std::string> received_types_;
Comment thread
juanlofer-eprosima marked this conversation as resolved.

void notify_type_discovered_(
const fastdds::dds::xtypes::TypeInformation& type_info,
const std::string& type_name);

};

protected:

void notify_type_discovered_(
const fastdds::dds::xtypes::TypeInformation& type_info,
const std::string& type_name);
//! Override method from \c CommonParticipant to create the internal RTPS participant listener
std::unique_ptr<fastdds::rtps::RTPSParticipantListener> create_listener_() override;

//! Type Object Internal Reader
std::shared_ptr<InternalReader> type_object_reader_;
Comment thread
juanlofer-eprosima marked this conversation as resolved.

//! Received types set
std::set<std::string> received_types_;
};

} /* namespace participants */
Expand Down
101 changes: 65 additions & 36 deletions ddspipe_participants/include/ddspipe_participants/participant/rtps/CommonParticipant.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@ namespace rtps {
*/
class CommonParticipant
: public core::IParticipant
, public fastdds::rtps::RTPSParticipantListener
{
public:

Expand Down Expand Up @@ -123,41 +122,62 @@ class CommonParticipant
// RTPS LISTENER METHODS
/////////////////////////

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method only is for debugging purposes.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_participant_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ParticipantDiscoveryStatus reason,
const fastdds::rtps::ParticipantBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method adds to database the endpoint discovered or modified.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_reader_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::rtps::SubscriptionBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method adds to database the endpoint discovered or modified.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_writer_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::rtps::PublicationBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;
class RtpsListener : public fastdds::rtps::RTPSParticipantListener
{
public:

DDSPIPE_PARTICIPANTS_DllAPI
explicit RtpsListener(
std::shared_ptr<ParticipantConfiguration> conf,
std::shared_ptr<core::DiscoveryDatabase> ddb);

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method is only used for debugging purposes.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_participant_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ParticipantDiscoveryStatus reason,
const fastdds::rtps::ParticipantBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method adds to database the endpoint discovered or modified.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_reader_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::ReaderDiscoveryStatus reason,
const fastdds::rtps::SubscriptionBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

/**
* @brief Override method from \c RTPSParticipantListener .
*
* This method adds to database the endpoint discovered or modified.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual void on_writer_discovery(
fastdds::rtps::RTPSParticipant* participant,
fastdds::rtps::WriterDiscoveryStatus reason,
const fastdds::rtps::PublicationBuiltinTopicData& info,
bool& /*should_be_ignored*/) override;

protected:

//! Shared pointer to the configuration of the participant
const std::shared_ptr<ParticipantConfiguration> configuration_;
//! Shared pointer to the discovery database
const std::shared_ptr<core::DiscoveryDatabase> discovery_database_;

};

//! Unique pointer to the internal RTPS Participant Listener
std::unique_ptr<fastdds::rtps::RTPSParticipantListener> rtps_participant_listener_;

//////////////////
// STATIC METHODS
Expand Down Expand Up @@ -221,6 +241,15 @@ class CommonParticipant
DDSPIPE_PARTICIPANTS_DllAPI
virtual fastdds::rtps::RTPSParticipantAttributes reckon_participant_attributes_() const;

/**
* @brief Virtual method that creates a listener for the internal RTPS Participant.
* It should be overridden if a different listener is needed.
* This method must be called after the RTPS Participant is created, otherwise no listener will be set.
* @return A unique pointer to an RTPS Participant Listener.
*/
DDSPIPE_PARTICIPANTS_DllAPI
virtual std::unique_ptr<fastdds::rtps::RTPSParticipantListener> create_listener_();

/////
// VARIABLES

Expand Down
Loading