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.

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

Jump to bottom

[19696][20409] Fix data race on PDP (backport #4220) #4431

Merged
merged 2 commits into from
Mar 22, 2024
Merged

[19696][20409] Fix data race on PDP (backport #4220) #4431

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0d1cfd0
Fix data race on PDP (#4220)
juanlofer-eprosima Feb 23, 2024
5995476
Export ParticipantProxyData constructor/destructor
juanlofer-eprosima Feb 29, 2024
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
7 changes: 3 additions & 4 deletions include/fastdds/rtps/attributes/ServerAttributes.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,15 +54,14 @@ class RemoteServerAttributes
return guidPrefix == r.guidPrefix
&& metatrafficUnicastLocatorList == r.metatrafficUnicastLocatorList
&& metatrafficMulticastLocatorList == r.metatrafficMulticastLocatorList;
// && proxy == r.proxy;
}

RTPS_DllAPI void clear()
{
guidPrefix = fastrtps::rtps::GuidPrefix_t::unknown();
metatrafficUnicastLocatorList.clear();
metatrafficMulticastLocatorList.clear();
proxy = nullptr;
is_connected = false;
}

RTPS_DllAPI fastrtps::rtps::GUID_t GetParticipant() const;
Expand Down Expand Up @@ -100,8 +99,8 @@ class RemoteServerAttributes
//!Guid prefix
fastrtps::rtps::GuidPrefix_t guidPrefix;

// Live participant proxy reference
const fastrtps::rtps::ParticipantProxyData* proxy{};
// Whether connection has been established
bool is_connected = false;

// Check if there are specific transport locators associated
// the template parameter is the locator kind (e.g. LOCATOR_KIND_UDPv4)
Expand Down
6 changes: 3 additions & 3 deletions include/fastdds/rtps/builtin/data/ParticipantProxyData.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,13 @@ class ParticipantProxyData
{
public:

ParticipantProxyData(
RTPS_DllAPI ParticipantProxyData(
const RTPSParticipantAllocationAttributes& allocation);

ParticipantProxyData(
RTPS_DllAPI ParticipantProxyData(
const ParticipantProxyData& pdata);

virtual ~ParticipantProxyData();
RTPS_DllAPI virtual ~ParticipantProxyData();

//!Protocol version
ProtocolVersion_t m_protocolVersion;
Expand Down
20 changes: 9 additions & 11 deletions src/cpp/rtps/builtin/discovery/participant/PDPClient.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,7 +459,6 @@ bool PDPClient::create_ds_pdp_reliable_endpoints(
return true;
}

// the ParticipantProxyData* pdata must be the one kept in PDP database
void PDPClient::assignRemoteEndpoints(
ParticipantProxyData* pdata)
{
Expand All @@ -475,8 +474,7 @@ void PDPClient::assignRemoteEndpoints(
{
if (data_matches_with_prefix(svr.guidPrefix, *pdata))
{
std::unique_lock<std::recursive_mutex> lock(*getMutex());
svr.proxy = pdata;
svr.is_connected = true;
}
}
}
Expand Down Expand Up @@ -504,11 +502,11 @@ void PDPClient::notifyAboveRemoteEndpoints(
{
if (data_matches_with_prefix(svr.guidPrefix, pdata))
{
if (nullptr == svr.proxy)
if (!svr.is_connected && nullptr != get_participant_proxy_data(svr.guidPrefix))
{
//! try to retrieve the participant proxy data from an unmangled prefix in case
//! we could not fill svr.proxy in assignRemoteEndpoints()
svr.proxy = get_participant_proxy_data(svr.guidPrefix);
//! mark proxy as connected from an unmangled prefix in case
//! it could not be done in assignRemoteEndpoints()
svr.is_connected = true;
}

match_pdp_reader_nts_(svr, pdata.m_guid.guidPrefix);
Expand Down Expand Up @@ -583,7 +581,7 @@ void PDPClient::removeRemoteEndpoints(
if (svr.guidPrefix == pdata->m_guid.guidPrefix)
{
std::unique_lock<std::recursive_mutex> lock(*getMutex());
svr.proxy = nullptr; // reasign when we receive again server DATA(p)
svr.is_connected = false;
is_server = true;
mp_sync->restart_timer(); // enable announcement and sync mechanism till this server reappears
}
Expand Down Expand Up @@ -755,11 +753,11 @@ void PDPClient::announceParticipantState(
for (auto& svr : mp_builtin->m_DiscoveryServers)
{
// if we are matched to a server report demise
if (svr.proxy != nullptr)
if (svr.is_connected)
{
//locators.push_back(svr.metatrafficMulticastLocatorList);
locators.push_back(svr.metatrafficUnicastLocatorList);
remote_readers.emplace_back(svr.proxy->m_guid.guidPrefix,
remote_readers.emplace_back(svr.guidPrefix,
endpoints->reader.reader_->getGuid().entityId);
}
}
Expand Down Expand Up @@ -792,7 +790,7 @@ void PDPClient::announceParticipantState(
{
// non-pinging announcements like lease duration ones must be
// broadcast to all servers
if (svr.proxy == nullptr || !_serverPing)
if (!svr.is_connected || !_serverPing)
{
locators.push_back(svr.metatrafficMulticastLocatorList);
locators.push_back(svr.metatrafficUnicastLocatorList);
Expand Down
22 changes: 17 additions & 5 deletions src/cpp/rtps/builtin/discovery/participant/PDPListener.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,11 +153,14 @@ void PDPListener::process_alive_data(
// Create a new one when not found
old_data = parent_pdp_->createParticipantProxyData(new_data, writer_guid);

reader->getMutex().unlock();
lock.unlock();

if (old_data != nullptr)
{
// Copy proxy to be passed forward before releasing PDP mutex
ParticipantProxyData old_data_copy(*old_data);

reader->getMutex().unlock();
lock.unlock();

// Assigning remote endpoints implies sending a DATA(p) to all matched and fixed readers, since
// StatelessWriter::matched_reader_add marks the entire history as unsent if the added reader's
// durability is bigger or equal to TRANSIENT_LOCAL_DURABILITY_QOS (TRANSIENT_LOCAL or TRANSIENT),
Expand All @@ -168,13 +171,19 @@ void PDPListener::process_alive_data(
// participant is discovered in the middle of BuiltinProtocols::initBuiltinProtocols, which will
// create the first DATA(p) upon finishing, thus triggering the sent to all fixed and matched
// readers anyways.
parent_pdp_->assignRemoteEndpoints(old_data);
parent_pdp_->assignRemoteEndpoints(&old_data_copy);
}
else
{
reader->getMutex().unlock();
lock.unlock();
}
}
else
{
old_data->updateData(new_data);
old_data->isAlive = true;

reader->getMutex().unlock();

logInfo(RTPS_PDP_DISCOVERY, "Update participant "
Expand All @@ -187,14 +196,17 @@ void PDPListener::process_alive_data(
parent_pdp_->mp_EDP->assignRemoteEndpoints(*old_data, true);
}

// Copy proxy to be passed forward before releasing PDP mutex
ParticipantProxyData old_data_copy(*old_data);

lock.unlock();

RTPSParticipantListener* listener = parent_pdp_->getRTPSParticipant()->getListener();
if (listener != nullptr)
{
{
std::lock_guard<std::mutex> cb_lock(parent_pdp_->callback_mtx_);
ParticipantDiscoveryInfo info(*old_data);
ParticipantDiscoveryInfo info(old_data_copy);
info.status = ParticipantDiscoveryInfo::CHANGED_QOS_PARTICIPANT;

listener->onParticipantDiscovery(
Expand Down
Loading
Loading