Skip to content

Commit feb439a

Browse files
MiguelCompanycferreiragonz
MiguelCompany
authored and
cferreiragonz
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> (cherry picked from commit de70d42) # Conflicts: # test/blackbox/common/DDSBlackboxTestsDiscovery.cpp
1 parent ab1631c commit feb439a

6 files changed

Lines changed: 316 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
@@ -1320,6 +1320,8 @@ void DataWriterImpl::InnerDataWriterListener::on_writer_matched(
13201320
RTPSWriter* /*writer*/,
13211321
const MatchingInfo& info)
13221322
{
1323+
std::lock_guard<std::mutex> scoped_lock(matching_info_mutex_);
1324+
13231325
data_writer_->update_publication_matched_status(info);
13241326

13251327
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
@@ -487,7 +487,10 @@ class DataWriterImpl : protected rtps::IReaderDataFilter
487487
const uint32_t& status_id);
488488
#endif //FASTDDS_STATISTICS
489489

490+
private:
491+
490492
DataWriterImpl* data_writer_;
493+
std::mutex matching_info_mutex_;
491494
}
492495
writer_listener_;
493496

‎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
@@ -481,7 +481,10 @@ class DataReaderImpl
481481
const uint32_t& status_id);
482482
#endif //FASTDDS_STATISTICS
483483

484+
private:
485+
484486
DataReaderImpl* data_reader_;
487+
std::mutex matching_info_mutex_;
485488

486489
}
487490
reader_listener_;

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

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

800800
//MATCHED AND ADDED CORRECTLY:
801-
ReaderListener* listener = nullptr;
802-
if (nullptr != (listener = r.get_listener()))
801+
ReaderListener* listener = r.get_listener();
802+
if (nullptr != listener)
803803
{
804804
MatchingInfo info;
805805
info.status = REMOVED_MATCHING;
@@ -833,8 +833,8 @@ bool EDP::unpairReaderProxy(
833833
participant_guid, reader_guid);
834834
#endif // if HAVE_SECURITY
835835
//MATCHED AND ADDED CORRECTLY:
836-
WriterListener* listener = nullptr;
837-
if (nullptr != (listener = w.get_listener()))
836+
WriterListener* listener = w.get_listener();
837+
if (nullptr != listener)
838838
{
839839
MatchingInfo info;
840840
info.status = REMOVED_MATCHING;
@@ -1143,40 +1143,39 @@ bool EDP::pairingReader(
11431143
"WP:" << wdatait->guid << " match R:" << reader_guid << ". RLoc:" <<
11441144
wdatait->remote_locators);
11451145
//MATCHED AND ADDED CORRECTLY:
1146-
if (reader->get_listener() != nullptr)
1146+
ReaderListener* listener = reader->get_listener();
1147+
if (nullptr != listener)
11471148
{
11481149
MatchingInfo info;
11491150
info.status = MATCHED_MATCHING;
11501151
info.remoteEndpointGuid = writer_guid;
1151-
reader->get_listener()->on_reader_matched(reader, info);
1152+
listener->on_reader_matched(reader, info);
11521153
}
11531154
}
11541155
#endif // if HAVE_SECURITY
11551156
}
11561157
else
11571158
{
1158-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && reader->get_listener() != nullptr)
1159+
ReaderListener* listener = reader->get_listener();
1160+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
11591161
{
1160-
reader->get_listener()->on_requested_incompatible_qos(reader, incompatible_qos);
1162+
listener->on_requested_incompatible_qos(reader, incompatible_qos);
11611163
mp_PDP->notify_incompatible_qos_matching(R->getGuid(), wdatait->guid, incompatible_qos);
11621164
}
11631165

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

11731172
//MATCHED AND ADDED CORRECTLY:
1174-
if (reader->get_listener() != nullptr)
1173+
if (nullptr != listener)
11751174
{
11761175
MatchingInfo info;
11771176
info.status = REMOVED_MATCHING;
11781177
info.remoteEndpointGuid = writer_guid;
1179-
reader->get_listener()->on_reader_matched(reader, info);
1178+
listener->on_reader_matched(reader, info);
11801179
}
11811180
}
11821181
}
@@ -1236,21 +1235,23 @@ bool EDP::pairingWriter(
12361235
"RP:" << rdatait->guid << " match W:" << writer_guid << ". WLoc:" <<
12371236
rdatait->remote_locators);
12381237
//MATCHED AND ADDED CORRECTLY:
1239-
if (writer->get_listener() != nullptr)
1238+
WriterListener* listener = writer->get_listener();
1239+
if (nullptr != listener)
12401240
{
12411241
MatchingInfo info;
12421242
info.status = MATCHED_MATCHING;
12431243
info.remoteEndpointGuid = reader_guid;
1244-
writer->get_listener()->on_writer_matched(writer, info);
1244+
listener->on_writer_matched(writer, info);
12451245
}
12461246
}
12471247
#endif // if HAVE_SECURITY
12481248
}
12491249
else
12501250
{
1251-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && writer->get_listener() != nullptr)
1251+
WriterListener* listener = writer->get_listener();
1252+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
12521253
{
1253-
writer->get_listener()->on_offered_incompatible_qos(writer, incompatible_qos);
1254+
listener->on_offered_incompatible_qos(writer, incompatible_qos);
12541255
mp_PDP->notify_incompatible_qos_matching(W->getGuid(), rdatait->guid, incompatible_qos);
12551256
}
12561257

@@ -1261,12 +1262,12 @@ bool EDP::pairingWriter(
12611262
mp_RTPSParticipant->security_manager().remove_reader(writer_guid, participant_guid, reader_guid);
12621263
#endif // if HAVE_SECURITY
12631264
//MATCHED AND ADDED CORRECTLY:
1264-
if (writer->get_listener() != nullptr)
1265+
if (nullptr != listener)
12651266
{
12661267
MatchingInfo info;
12671268
info.status = REMOVED_MATCHING;
12681269
info.remoteEndpointGuid = reader_guid;
1269-
writer->get_listener()->on_writer_matched(writer, info);
1270+
listener->on_writer_matched(writer, info);
12701271
}
12711272
}
12721273
}
@@ -1312,38 +1313,39 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(
13121313
"RP:" << rdata->guid << " match W:" << w.getGuid() << ". RLoc:" <<
13131314
rdata->remote_locators);
13141315
//MATCHED AND ADDED CORRECTLY:
1315-
if (w.get_listener() != nullptr)
1316+
WriterListener* listener = w.get_listener();
1317+
if (nullptr != listener)
13161318
{
13171319
MatchingInfo info;
13181320
info.status = MATCHED_MATCHING;
13191321
info.remoteEndpointGuid = reader_guid;
1320-
w.get_listener()->on_writer_matched(&w, info);
1322+
listener->on_writer_matched(&w, info);
13211323
}
13221324
}
13231325
#endif // if HAVE_SECURITY
13241326
}
13251327
else
13261328
{
1327-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && w.get_listener() != nullptr)
1329+
WriterListener* listener = w.get_listener();
1330+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
13281331
{
1329-
w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos);
1332+
listener->on_offered_incompatible_qos(&w, incompatible_qos);
13301333
mp_PDP->notify_incompatible_qos_matching(w.getGuid(), rdata->guid, incompatible_qos);
13311334
}
13321335

1333-
if (w.matched_reader_is_matched(reader_guid)
1334-
&& w.matched_reader_remove(reader_guid))
1336+
if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid))
13351337
{
13361338
#if HAVE_SECURITY
13371339
mp_RTPSParticipant->security_manager().remove_reader(
13381340
w.getGuid(), participant_guid, reader_guid);
13391341
#endif // if HAVE_SECURITY
13401342
//MATCHED AND ADDED CORRECTLY:
1341-
if (w.get_listener() != nullptr)
1343+
if (nullptr != listener)
13421344
{
13431345
MatchingInfo info;
13441346
info.status = REMOVED_MATCHING;
13451347
info.remoteEndpointGuid = reader_guid;
1346-
w.get_listener()->on_writer_matched(&w, info);
1348+
listener->on_writer_matched(&w, info);
13471349
}
13481350
}
13491351
}
@@ -1392,25 +1394,24 @@ bool EDP::pairing_reader_proxy_with_local_writer(
13921394
}
13931395
else
13941396
{
1395-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
1396-
w.get_listener() != nullptr)
1397+
WriterListener* listener = w.get_listener();
1398+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
13971399
{
1398-
w.get_listener()->on_offered_incompatible_qos(&w, incompatible_qos);
1400+
listener->on_offered_incompatible_qos(&w, incompatible_qos);
13991401
mp_PDP->notify_incompatible_qos_matching(local_writer, rdata.guid, incompatible_qos);
14001402
}
14011403

1402-
if (w.matched_reader_is_matched(reader_guid)
1403-
&& w.matched_reader_remove(reader_guid))
1404+
if (w.matched_reader_is_matched(reader_guid) && w.matched_reader_remove(reader_guid))
14041405
{
14051406
mp_RTPSParticipant->security_manager().remove_reader(w.getGuid(),
14061407
remote_participant_guid, reader_guid);
14071408
//MATCHED AND ADDED CORRECTLY:
1408-
if (w.get_listener() != nullptr)
1409+
if (nullptr != listener)
14091410
{
14101411
MatchingInfo info;
14111412
info.status = REMOVED_MATCHING;
14121413
info.remoteEndpointGuid = reader_guid;
1413-
w.get_listener()->on_writer_matched(&w, info);
1414+
listener->on_writer_matched(&w, info);
14141415
}
14151416
}
14161417
}
@@ -1447,12 +1448,13 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security(
14471448
matched = true;
14481449

14491450
//MATCHED AND ADDED CORRECTLY:
1450-
if (w.get_listener() != nullptr)
1451+
WriterListener* listener = w.get_listener();
1452+
if (nullptr != listener)
14511453
{
14521454
MatchingInfo info;
14531455
info.status = MATCHED_MATCHING;
14541456
info.remoteEndpointGuid = reader_guid;
1455-
w.get_listener()->on_writer_matched(&w, info);
1457+
listener->on_writer_matched(&w, info);
14561458
}
14571459
}
14581460
// don't look anymore
@@ -1505,38 +1507,39 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(
15051507
"WP:" << wdata->guid << " match R:" << r.getGuid() << ". WLoc:" <<
15061508
wdata->remote_locators);
15071509
//MATCHED AND ADDED CORRECTLY:
1508-
if (r.get_listener() != nullptr)
1510+
ReaderListener* listener = r.get_listener();
1511+
if (nullptr != listener)
15091512
{
15101513
MatchingInfo info;
15111514
info.status = MATCHED_MATCHING;
15121515
info.remoteEndpointGuid = writer_guid;
1513-
r.get_listener()->on_reader_matched(&r, info);
1516+
listener->on_reader_matched(&r, info);
15141517
}
15151518
}
15161519
#endif // if HAVE_SECURITY
15171520
}
15181521
else
15191522
{
1520-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && r.get_listener() != nullptr)
1523+
ReaderListener* listener = r.get_listener();
1524+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
15211525
{
1522-
r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos);
1526+
listener->on_requested_incompatible_qos(&r, incompatible_qos);
15231527
mp_PDP->notify_incompatible_qos_matching(r.getGuid(), wdata->guid, incompatible_qos);
15241528
}
15251529

1526-
if (r.matched_writer_is_matched(writer_guid)
1527-
&& r.matched_writer_remove(writer_guid))
1530+
if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid))
15281531
{
15291532
#if HAVE_SECURITY
15301533
mp_RTPSParticipant->security_manager().remove_writer(readerGUID, participant_guid,
15311534
writer_guid);
15321535
#endif // if HAVE_SECURITY
15331536
//MATCHED AND ADDED CORRECTLY:
1534-
if (r.get_listener() != nullptr)
1537+
if (nullptr != listener)
15351538
{
15361539
MatchingInfo info;
15371540
info.status = REMOVED_MATCHING;
15381541
info.remoteEndpointGuid = writer_guid;
1539-
r.get_listener()->on_reader_matched(&r, info);
1542+
listener->on_reader_matched(&r, info);
15401543
}
15411544
}
15421545
}
@@ -1585,25 +1588,24 @@ bool EDP::pairing_writer_proxy_with_local_reader(
15851588
}
15861589
else
15871590
{
1588-
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) &&
1589-
r.get_listener() != nullptr)
1591+
ReaderListener* listener = r.get_listener();
1592+
if (no_match_reason.test(MatchingFailureMask::incompatible_qos) && (nullptr != listener))
15901593
{
1591-
r.get_listener()->on_requested_incompatible_qos(&r, incompatible_qos);
1594+
listener->on_requested_incompatible_qos(&r, incompatible_qos);
15921595
mp_PDP->notify_incompatible_qos_matching(local_reader, wdata.guid, incompatible_qos);
15931596
}
15941597

1595-
if (r.matched_writer_is_matched(writer_guid)
1596-
&& r.matched_writer_remove(writer_guid))
1598+
if (r.matched_writer_is_matched(writer_guid) && r.matched_writer_remove(writer_guid))
15971599
{
15981600
mp_RTPSParticipant->security_manager().remove_writer(readerGUID,
15991601
remote_participant_guid, writer_guid);
16001602
//MATCHED AND ADDED CORRECTLY:
1601-
if (r.get_listener() != nullptr)
1603+
if (nullptr != listener)
16021604
{
16031605
MatchingInfo info;
16041606
info.status = REMOVED_MATCHING;
16051607
info.remoteEndpointGuid = writer_guid;
1606-
r.get_listener()->on_reader_matched(&r, info);
1608+
listener->on_reader_matched(&r, info);
16071609
}
16081610
}
16091611
}
@@ -1642,13 +1644,13 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security(
16421644
matched = true;
16431645

16441646
//MATCHED AND ADDED CORRECTLY:
1645-
if (r.get_listener() != nullptr)
1647+
ReaderListener* listener = r.get_listener();
1648+
if (nullptr != listener)
16461649
{
16471650
MatchingInfo info;
16481651
info.status = MATCHED_MATCHING;
16491652
info.remoteEndpointGuid = writer_guid;
1650-
r.get_listener()->on_reader_matched(&r, info);
1651-
1653+
listener->on_reader_matched(&r, info);
16521654
}
16531655
}
16541656
// dont' look anymore

0 commit comments

Comments
 (0)