Skip to content

Commit eda2589

Browse files
MiguelCompanyMiguelCompany
committed
Make on_xxx_matched thread-safe (#6371)
* Refs #21355. Added regression test. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21355. Improve readability in EDP.cpp. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21355. Protect `InnerDataWriterListener::on_writer_matched`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> * Refs #21355. Protect `InnerDataReaderListener::on_reader_matched`. Signed-off-by: Miguel Company <miguelcompany@eprosima.com> --------- Signed-off-by: Miguel Company <miguelcompany@eprosima.com>
1 parent 5ada665 commit eda2589

6 files changed

Lines changed: 187 additions & 56 deletions

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1341,6 +1341,8 @@ void DataWriterImpl::InnerDataWriterListener::on_writer_matched(
13411341
RTPSWriter* /*writer*/,
13421342
const MatchingInfo& info)
13431343
{
1344+
std::lock_guard<std::mutex> scoped_lock(matching_info_mutex_);
1345+
13441346
data_writer_->update_publication_matched_status(info);
13451347

13461348
StatusMask notify_status = StatusMask::publication_matched();

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -520,7 +520,10 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
520520
const uint32_t& status_id);
521521
#endif //FASTDDS_STATISTICS
522522

523+
private:
524+
523525
DataWriterImpl* data_writer_;
526+
std::mutex matching_info_mutex_;
524527
}
525528
writer_listener_;
526529

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -941,6 +941,8 @@ void DataReaderImpl::InnerDataReaderListener::on_reader_matched(
941941
RTPSReader* /*reader*/,
942942
const MatchingInfo& info)
943943
{
944+
std::lock_guard<std::mutex> scoped_lock(matching_info_mutex_);
945+
944946
data_reader_->update_subscription_matched_status(info);
945947

946948
StatusMask notify_status = StatusMask::subscription_matched();

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -500,7 +500,10 @@ class DataReaderImpl
500500
const uint32_t& status_id);
501501
#endif //FASTDDS_STATISTICS
502502

503+
private:
504+
503505
DataReaderImpl* data_reader_;
506+
std::mutex matching_info_mutex_;
504507

505508
}
506509
reader_listener_;

‎src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp‎

Lines changed: 58 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -799,8 +799,8 @@ bool EDP::unpairWriterProxy(
799799
#endif // if HAVE_SECURITY
800800

801801
//MATCHED AND ADDED CORRECTLY:
802-
ReaderListener* listener = nullptr;
803-
if (nullptr != (listener = r.get_listener()))
802+
ReaderListener* listener = r.get_listener();
803+
if (nullptr != listener)
804804
{
805805
MatchingInfo info;
806806
info.status = REMOVED_MATCHING;
@@ -834,8 +834,8 @@ bool EDP::unpairReaderProxy(
834834
participant_guid, reader_guid);
835835
#endif // if HAVE_SECURITY
836836
//MATCHED AND ADDED CORRECTLY:
837-
WriterListener* listener = nullptr;
838-
if (nullptr != (listener = w.get_listener()))
837+
WriterListener* listener = w.get_listener();
838+
if (nullptr != listener)
839839
{
840840
MatchingInfo info;
841841
info.status = REMOVED_MATCHING;
@@ -1144,40 +1144,39 @@ bool EDP::pairingReader(
11441144
"WP:" << wdatait->guid << " match R:" << reader_guid << ". RLoc:" <<
11451145
wdatait->remote_locators);
11461146
//MATCHED AND ADDED CORRECTLY:
1147-
if (reader->get_listener() != nullptr)
1147+
ReaderListener* listener = reader->get_listener();
1148+
if (nullptr != listener)
11481149
{
11491150
MatchingInfo info;
11501151
info.status = MATCHED_MATCHING;
11511152
info.remoteEndpointGuid = writer_guid;
1152-
reader->get_listener()->on_reader_matched(reader, info);
1153+
listener->on_reader_matched(reader, info);
11531154
}
11541155
}
11551156
#endif // if HAVE_SECURITY
11561157
}
11571158
else
11581159
{
1159-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && reader->get_listener() != nullptr)
1160+
ReaderListener* listener = reader->get_listener();
1161+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
11601162
{
1161-
reader->get_listener()->on_requested_incompatible_qos(reader, incompatible_qos);
1163+
listener->on_requested_incompatible_qos(reader, incompatible_qos);
11621164
mp_PDP->notify_incompatible_qos_matching(R->getGuid(), wdatait->guid, incompatible_qos);
11631165
}
11641166

1165-
//EPROSIMA_LOG_INFO(RTPS_EDP,RTPS_CYAN<<"Valid Matching to writerProxy: "<<wdatait->guid<<RTPS_DEF<<endl);
1166-
if (reader->matched_writer_is_matched(wdatait->guid)
1167-
&& reader->matched_writer_remove(wdatait->guid))
1167+
if (reader->matched_writer_is_matched(wdatait->guid) && reader->matched_writer_remove(wdatait->guid))
11681168
{
11691169
#if HAVE_SECURITY
1170-
mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid,
1171-
wdatait->guid);
1170+
mp_RTPSParticipant->security_manager().remove_writer(reader_guid, participant_guid, wdatait->guid);
11721171
#endif // if HAVE_SECURITY
11731172

11741173
//MATCHED AND ADDED CORRECTLY:
1175-
if (reader->get_listener() != nullptr)
1174+
if (nullptr != listener)
11761175
{
11771176
MatchingInfo info;
11781177
info.status = REMOVED_MATCHING;
11791178
info.remoteEndpointGuid = writer_guid;
1180-
reader->get_listener()->on_reader_matched(reader, info);
1179+
listener->on_reader_matched(reader, info);
11811180
}
11821181
}
11831182
}
@@ -1237,21 +1236,23 @@ bool EDP::pairingWriter(
12371236
"RP:" << rdatait->guid << " match W:" << writer_guid << ". WLoc:" <<
12381237
rdatait->remote_locators);
12391238
//MATCHED AND ADDED CORRECTLY:
1240-
if (writer->get_listener() != nullptr)
1239+
WriterListener* listener = writer->get_listener();
1240+
if (nullptr != listener)
12411241
{
12421242
MatchingInfo info;
12431243
info.status = MATCHED_MATCHING;
12441244
info.remoteEndpointGuid = reader_guid;
1245-
writer->get_listener()->on_writer_matched(writer, info);
1245+
listener->on_writer_matched(writer, info);
12461246
}
12471247
}
12481248
#endif // if HAVE_SECURITY
12491249
}
12501250
else
12511251
{
1252-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && writer->get_listener() != nullptr)
1252+
WriterListener* listener = writer->get_listener();
1253+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
12531254
{
1254-
writer->get_listener()->on_offered_incompatible_qos(writer, incompatible_qos);
1255+
listener->on_offered_incompatible_qos(writer, incompatible_qos);
12551256
mp_PDP->notify_incompatible_qos_matching(W->getGuid(), rdatait->guid, incompatible_qos);
12561257
}
12571258

@@ -1262,12 +1263,12 @@ bool EDP::pairingWriter(
12621263
mp_RTPSParticipant->security_manager().remove_reader(writer_guid, participant_guid, reader_guid);
12631264
#endif // if HAVE_SECURITY
12641265
//MATCHED AND ADDED CORRECTLY:
1265-
if (writer->get_listener() != nullptr)
1266+
if (nullptr != listener)
12661267
{
12671268
MatchingInfo info;
12681269
info.status = REMOVED_MATCHING;
12691270
info.remoteEndpointGuid = reader_guid;
1270-
writer->get_listener()->on_writer_matched(writer, info);
1271+
listener->on_writer_matched(writer, info);
12711272
}
12721273
}
12731274
}
@@ -1313,38 +1314,39 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(
13131314
"RP:" << rdata->guid << " match W:" << w.getGuid() << ". RLoc:" <<
13141315
rdata->remote_locators);
13151316
//MATCHED AND ADDED CORRECTLY:
1316-
if (w.get_listener() != nullptr)
1317+
WriterListener* listener = w.get_listener();
1318+
if (nullptr != listener)
13171319
{
13181320
MatchingInfo info;
13191321
info.status = MATCHED_MATCHING;
13201322
info.remoteEndpointGuid = reader_guid;
1321-
w.get_listener()->on_writer_matched(&w, info);
1323+
listener->on_writer_matched(&w, info);
13221324
}
13231325
}
13241326
#endif // if HAVE_SECURITY
13251327
}
13261328
else
13271329
{
1328-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && w.get_listener() != nullptr)
1330+
WriterListener* listener = w.get_listener();
1331+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
13291332
{
1330-
w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos);
1333+
listener->on_offered_incompatible_qos(&w, incompatible_qos);
13311334
mp_PDP->notify_incompatible_qos_matching(w.getGuid(), rdata->guid, incompatible_qos);
13321335
}
13331336

1334-
if (w.matched_reader_is_matched(reader_guid)
1335-
&& w.matched_reader_remove(reader_guid))
1337+
if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid))
13361338
{
13371339
#if HAVE_SECURITY
13381340
mp_RTPSParticipant->security_manager().remove_reader(
13391341
w.getGuid(), participant_guid, reader_guid);
13401342
#endif // if HAVE_SECURITY
13411343
//MATCHED AND ADDED CORRECTLY:
1342-
if (w.get_listener() != nullptr)
1344+
if (nullptr != listener)
13431345
{
13441346
MatchingInfo info;
13451347
info.status = REMOVED_MATCHING;
13461348
info.remoteEndpointGuid = reader_guid;
1347-
w.get_listener()->on_writer_matched(&w, info);
1349+
listener->on_writer_matched(&w, info);
13481350
}
13491351
}
13501352
}
@@ -1393,25 +1395,24 @@ bool EDP::pairing_reader_proxy_with_local_writer(
13931395
}
13941396
else
13951397
{
1396-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
1397-
w.get_listener() != nullptr)
1398+
WriterListener* listener = w.get_listener();
1399+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
13981400
{
1399-
w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos);
1401+
listener->on_offered_incompatible_qos(&w, incompatible_qos);
14001402
mp_PDP->notify_incompatible_qos_matching(local_writer, rdata.guid, incompatible_qos);
14011403
}
14021404

1403-
if (w.matched_reader_is_matched(reader_guid)
1404-
&& w.matched_reader_remove(reader_guid))
1405+
if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid))
14051406
{
14061407
mp_RTPSParticipant->security_manager().remove_reader(w.getGuid(),
14071408
remote_participant_guid, reader_guid);
14081409
//MATCHED AND ADDED CORRECTLY:
1409-
if (w.get_listener() != nullptr)
1410+
if (nullptr != listener)
14101411
{
14111412
MatchingInfo info;
14121413
info.status = REMOVED_MATCHING;
14131414
info.remoteEndpointGuid = reader_guid;
1414-
w.get_listener()->on_writer_matched(&w, info);
1415+
listener->on_writer_matched(&w, info);
14151416
}
14161417
}
14171418
}
@@ -1448,12 +1449,13 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security(
14481449
matched = true;
14491450

14501451
//MATCHED AND ADDED CORRECTLY:
1451-
if (w.get_listener() != nullptr)
1452+
WriterListener* listener = w.get_listener();
1453+
if (nullptr != listener)
14521454
{
14531455
MatchingInfo info;
14541456
info.status = MATCHED_MATCHING;
14551457
info.remoteEndpointGuid = reader_guid;
1456-
w.get_listener()->on_writer_matched(&w, info);
1458+
listener->on_writer_matched(&w, info);
14571459
}
14581460
}
14591461
// don't look anymore
@@ -1506,38 +1508,39 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(
15061508
"WP:" << wdata->guid << " match R:" << r.getGuid() << ". WLoc:" <<
15071509
wdata->remote_locators);
15081510
//MATCHED AND ADDED CORRECTLY:
1509-
if (r.get_listener() != nullptr)
1511+
ReaderListener* listener = r.get_listener();
1512+
if (nullptr != listener)
15101513
{
15111514
MatchingInfo info;
15121515
info.status = MATCHED_MATCHING;
15131516
info.remoteEndpointGuid = writer_guid;
1514-
r.get_listener()->on_reader_matched(&r, info);
1517+
listener->on_reader_matched(&r, info);
15151518
}
15161519
}
15171520
#endif // if HAVE_SECURITY
15181521
}
15191522
else
15201523
{
1521-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && r.get_listener() != nullptr)
1524+
ReaderListener* listener = r.get_listener();
1525+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
15221526
{
1523-
r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos);
1527+
listener->on_requested_incompatible_qos(&r, incompatible_qos);
15241528
mp_PDP->notify_incompatible_qos_matching(r.getGuid(), wdata->guid, incompatible_qos);
15251529
}
15261530

1527-
if (r.matched_writer_is_matched(writer_guid)
1528-
&& r.matched_writer_remove(writer_guid))
1531+
if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid))
15291532
{
15301533
#if HAVE_SECURITY
15311534
mp_RTPSParticipant->security_manager().remove_writer(readerGUID, participant_guid,
15321535
writer_guid);
15331536
#endif // if HAVE_SECURITY
15341537
//MATCHED AND ADDED CORRECTLY:
1535-
if (r.get_listener() != nullptr)
1538+
if (nullptr != listener)
15361539
{
15371540
MatchingInfo info;
15381541
info.status = REMOVED_MATCHING;
15391542
info.remoteEndpointGuid = writer_guid;
1540-
r.get_listener()->on_reader_matched(&r, info);
1543+
listener->on_reader_matched(&r, info);
15411544
}
15421545
}
15431546
}
@@ -1586,25 +1589,24 @@ bool EDP::pairing_writer_proxy_with_local_reader(
15861589
}
15871590
else
15881591
{
1589-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
1590-
r.get_listener() != nullptr)
1592+
ReaderListener* listener = r.get_listener();
1593+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
15911594
{
1592-
r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos);
1595+
listener->on_requested_incompatible_qos(&r, incompatible_qos);
15931596
mp_PDP->notify_incompatible_qos_matching(local_reader, wdata.guid, incompatible_qos);
15941597
}
15951598

1596-
if (r.matched_writer_is_matched(writer_guid)
1597-
&& r.matched_writer_remove(writer_guid))
1599+
if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid))
15981600
{
15991601
mp_RTPSParticipant->security_manager().remove_writer(readerGUID,
16001602
remote_participant_guid, writer_guid);
16011603
//MATCHED AND ADDED CORRECTLY:
1602-
if (r.get_listener() != nullptr)
1604+
if (nullptr != listener)
16031605
{
16041606
MatchingInfo info;
16051607
info.status = REMOVED_MATCHING;
16061608
info.remoteEndpointGuid = writer_guid;
1607-
r.get_listener()->on_reader_matched(&r, info);
1609+
listener->on_reader_matched(&r, info);
16081610
}
16091611
}
16101612
}
@@ -1643,13 +1645,13 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security(
16431645
matched = true;
16441646

16451647
//MATCHED AND ADDED CORRECTLY:
1646-
if (r.get_listener() != nullptr)
1648+
ReaderListener* listener = r.get_listener();
1649+
if (nullptr != listener)
16471650
{
16481651
MatchingInfo info;
16491652
info.status = MATCHED_MATCHING;
16501653
info.remoteEndpointGuid = writer_guid;
1651-
r.get_listener()->on_reader_matched(&r, info);
1652-
1654+
listener->on_reader_matched(&r, info);
16531655
}
16541656
}
16551657
// dont' look anymore

0 commit comments

Comments
 (0)