Refs #21355: Avoid dataraces on RTPS endpoint matching callbacks

Carlosespicur · Carlosespicur · commit 0591ddf9f5d7 · 2025-03-26T15:32:14.000+01:00
Signed-off-by: Carlosespicur &lt;carlosespicur@proton.me&gt;
diff --git a/src/cpp/fastdds/subscriber/DataReaderImpl.hpp b/src/cpp/fastdds/subscriber/DataReaderImpl.hpp
@@ -482,7 +482,6 @@ class DataReaderImpl
 #endif //FASTDDS_STATISTICS
 
         DataReaderImpl* data_reader_;
-
     }
     reader_listener_;
 
diff --git a/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp b/src/cpp/rtps/builtin/discovery/endpoint/EDP.cpp
@@ -524,6 +524,8 @@ bool EDP::unpairWriterProxy(
                         MatchingInfo info;
                         info.status = REMOVED_MATCHING;
                         info.remoteEndpointGuid = writer_guid;
+
+                        std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());
                         listener->on_reader_matched(&r, info);
                     }
                 }
@@ -559,6 +561,7 @@ bool EDP::unpairReaderProxy(
                         MatchingInfo info;
                         info.status = REMOVED_MATCHING;
                         info.remoteEndpointGuid = reader_guid;
+                        std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());
                         listener->on_writer_matched(&w, info);
                     }
                 }
@@ -868,6 +871,7 @@ bool EDP::pairingReader(
                         MatchingInfo info;
                         info.status = MATCHED_MATCHING;
                         info.remoteEndpointGuid = writer_guid;
+                        std::lock_guard<RecursiveTimedMutex> lock(reader->getMutex());
                         reader->get_listener()->on_reader_matched(reader, info);
                     }
                 }
@@ -896,6 +900,7 @@ bool EDP::pairingReader(
                         MatchingInfo info;
                         info.status = REMOVED_MATCHING;
                         info.remoteEndpointGuid = writer_guid;
+                        std::lock_guard<RecursiveTimedMutex> lock(reader->getMutex());
                         reader->get_listener()->on_reader_matched(reader, info);
                     }
                 }
@@ -961,6 +966,7 @@ bool EDP::pairingWriter(
                         MatchingInfo info;
                         info.status = MATCHED_MATCHING;
                         info.remoteEndpointGuid = reader_guid;
+                        std::lock_guard<RecursiveTimedMutex> lock(writer->getMutex());
                         writer->get_listener()->on_writer_matched(writer, info);
                     }
                 }
@@ -986,6 +992,7 @@ bool EDP::pairingWriter(
                         MatchingInfo info;
                         info.status = REMOVED_MATCHING;
                         info.remoteEndpointGuid = reader_guid;
+                        std::lock_guard<RecursiveTimedMutex> lock(writer->getMutex());
                         writer->get_listener()->on_writer_matched(writer, info);
                     }
                 }
@@ -1037,6 +1044,7 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(
                                 MatchingInfo info;
                                 info.status = MATCHED_MATCHING;
                                 info.remoteEndpointGuid = reader_guid;
+                                std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());
                                 w.get_listener()->on_writer_matched(&w, info);
                             }
                         }
@@ -1063,6 +1071,7 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(
                                 MatchingInfo info;
                                 info.status = REMOVED_MATCHING;
                                 info.remoteEndpointGuid = reader_guid;
+                                std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());
                                 w.get_listener()->on_writer_matched(&w, info);
                             }
                         }
@@ -1130,6 +1139,7 @@ bool EDP::pairing_reader_proxy_with_local_writer(
                                     MatchingInfo info;
                                     info.status = REMOVED_MATCHING;
                                     info.remoteEndpointGuid = reader_guid;
+                                    std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());
                                     w.get_listener()->on_writer_matched(&w, info);
                                 }
                             }
@@ -1172,6 +1182,7 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security(
                             MatchingInfo info;
                             info.status = MATCHED_MATCHING;
                             info.remoteEndpointGuid = reader_guid;
+                            std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());
                             w.get_listener()->on_writer_matched(&w, info);
                         }
                     }
@@ -1230,6 +1241,7 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(
                                 MatchingInfo info;
                                 info.status = MATCHED_MATCHING;
                                 info.remoteEndpointGuid = writer_guid;
+                                std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());
                                 r.get_listener()->on_reader_matched(&r, info);
                             }
                         }
@@ -1256,6 +1268,7 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(
                                 MatchingInfo info;
                                 info.status = REMOVED_MATCHING;
                                 info.remoteEndpointGuid = writer_guid;
+                                std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());
                                 r.get_listener()->on_reader_matched(&r, info);
                             }
                         }
@@ -1323,6 +1336,7 @@ bool EDP::pairing_writer_proxy_with_local_reader(
                                     MatchingInfo info;
                                     info.status = REMOVED_MATCHING;
                                     info.remoteEndpointGuid = writer_guid;
+                                    std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());
                                     r.get_listener()->on_reader_matched(&r, info);
                                 }
                             }
@@ -1367,6 +1381,7 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security(
                             MatchingInfo info;
                             info.status = MATCHED_MATCHING;
                             info.remoteEndpointGuid = writer_guid;
+                            std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());
                             r.get_listener()->on_reader_matched(&r, info);
 
                         }
diff --git a/test/dds/xtypes/TypeLookupServicePublisher.cpp b/test/dds/xtypes/TypeLookupServicePublisher.cpp
@@ -305,7 +305,6 @@ void TypeLookupServicePublisher::on_publication_matched(
         DataWriter* /*writer*/,
         const PublicationMatchedStatus& info)
 {
-    std::unique_lock<std::mutex> lock(mutex_);
     matched_ += info.current_count_change;
     cv_.notify_all();
 }
diff --git a/test/dds/xtypes/TypeLookupServiceSubscriber.cpp b/test/dds/xtypes/TypeLookupServiceSubscriber.cpp
@@ -364,7 +364,6 @@ void TypeLookupServiceSubscriber::on_subscription_matched(
         DataReader* /*reader*/,
         const SubscriptionMatchedStatus& info)
 {
-    std::unique_lock<std::mutex> lock(mutex_);
     matched_ += info.current_count_change;
     cv_.notify_all();
 }

Original file line number	Diff line number	Diff line change
`@@ -482,7 +482,6 @@ class DataReaderImpl`
`482`	`482`	`#endif //FASTDDS_STATISTICS`
`483`	`483`
`484`	`484`	`DataReaderImpl* data_reader_;`
`485`		`-`
`486`	`485`	`}`
`487`	`486`	`reader_listener_;`
`488`	`487`
Original file line number	Diff line number	Diff line change
`@@ -524,6 +524,8 @@ bool EDP::unpairWriterProxy(`
`524`	`524`	`MatchingInfo info;`
`525`	`525`	`info.status = REMOVED_MATCHING;`
`526`	`526`	`info.remoteEndpointGuid = writer_guid;`
	`527`	`+`
	`528`	`+ std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());`
`527`	`529`	`listener->on_reader_matched(&r, info);`
`528`	`530`	`}`
`529`	`531`	`}`
`@@ -559,6 +561,7 @@ bool EDP::unpairReaderProxy(`
`559`	`561`	`MatchingInfo info;`
`560`	`562`	`info.status = REMOVED_MATCHING;`
`561`	`563`	`info.remoteEndpointGuid = reader_guid;`
	`564`	`+ std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());`
`562`	`565`	`listener->on_writer_matched(&w, info);`
`563`	`566`	`}`
`564`	`567`	`}`
`@@ -868,6 +871,7 @@ bool EDP::pairingReader(`
`868`	`871`	`MatchingInfo info;`
`869`	`872`	`info.status = MATCHED_MATCHING;`
`870`	`873`	`info.remoteEndpointGuid = writer_guid;`
	`874`	`+ std::lock_guard<RecursiveTimedMutex> lock(reader->getMutex());`
`871`	`875`	`reader->get_listener()->on_reader_matched(reader, info);`
`872`	`876`	`}`
`873`	`877`	`}`
`@@ -896,6 +900,7 @@ bool EDP::pairingReader(`
`896`	`900`	`MatchingInfo info;`
`897`	`901`	`info.status = REMOVED_MATCHING;`
`898`	`902`	`info.remoteEndpointGuid = writer_guid;`
	`903`	`+ std::lock_guard<RecursiveTimedMutex> lock(reader->getMutex());`
`899`	`904`	`reader->get_listener()->on_reader_matched(reader, info);`
`900`	`905`	`}`
`901`	`906`	`}`
`@@ -961,6 +966,7 @@ bool EDP::pairingWriter(`
`961`	`966`	`MatchingInfo info;`
`962`	`967`	`info.status = MATCHED_MATCHING;`
`963`	`968`	`info.remoteEndpointGuid = reader_guid;`
	`969`	`+ std::lock_guard<RecursiveTimedMutex> lock(writer->getMutex());`
`964`	`970`	`writer->get_listener()->on_writer_matched(writer, info);`
`965`	`971`	`}`
`966`	`972`	`}`
`@@ -986,6 +992,7 @@ bool EDP::pairingWriter(`
`986`	`992`	`MatchingInfo info;`
`987`	`993`	`info.status = REMOVED_MATCHING;`
`988`	`994`	`info.remoteEndpointGuid = reader_guid;`
	`995`	`+ std::lock_guard<RecursiveTimedMutex> lock(writer->getMutex());`
`989`	`996`	`writer->get_listener()->on_writer_matched(writer, info);`
`990`	`997`	`}`
`991`	`998`	`}`
`@@ -1037,6 +1044,7 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(`
`1037`	`1044`	`MatchingInfo info;`
`1038`	`1045`	`info.status = MATCHED_MATCHING;`
`1039`	`1046`	`info.remoteEndpointGuid = reader_guid;`
	`1047`	`+ std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());`
`1040`	`1048`	`w.get_listener()->on_writer_matched(&w, info);`
`1041`	`1049`	`}`
`1042`	`1050`	`}`
`@@ -1063,6 +1071,7 @@ bool EDP::pairing_reader_proxy_with_any_local_writer(`
`1063`	`1071`	`MatchingInfo info;`
`1064`	`1072`	`info.status = REMOVED_MATCHING;`
`1065`	`1073`	`info.remoteEndpointGuid = reader_guid;`
	`1074`	`+ std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());`
`1066`	`1075`	`w.get_listener()->on_writer_matched(&w, info);`
`1067`	`1076`	`}`
`1068`	`1077`	`}`
`@@ -1130,6 +1139,7 @@ bool EDP::pairing_reader_proxy_with_local_writer(`
`1130`	`1139`	`MatchingInfo info;`
`1131`	`1140`	`info.status = REMOVED_MATCHING;`
`1132`	`1141`	`info.remoteEndpointGuid = reader_guid;`
	`1142`	`+ std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());`
`1133`	`1143`	`w.get_listener()->on_writer_matched(&w, info);`
`1134`	`1144`	`}`
`1135`	`1145`	`}`
`@@ -1172,6 +1182,7 @@ bool EDP::pairing_remote_reader_with_local_writer_after_security(`
`1172`	`1182`	`MatchingInfo info;`
`1173`	`1183`	`info.status = MATCHED_MATCHING;`
`1174`	`1184`	`info.remoteEndpointGuid = reader_guid;`
	`1185`	`+ std::lock_guard<RecursiveTimedMutex> lock(w.getMutex());`
`1175`	`1186`	`w.get_listener()->on_writer_matched(&w, info);`
`1176`	`1187`	`}`
`1177`	`1188`	`}`
`@@ -1230,6 +1241,7 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(`
`1230`	`1241`	`MatchingInfo info;`
`1231`	`1242`	`info.status = MATCHED_MATCHING;`
`1232`	`1243`	`info.remoteEndpointGuid = writer_guid;`
	`1244`	`+ std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());`
`1233`	`1245`	`r.get_listener()->on_reader_matched(&r, info);`
`1234`	`1246`	`}`
`1235`	`1247`	`}`
`@@ -1256,6 +1268,7 @@ bool EDP::pairing_writer_proxy_with_any_local_reader(`
`1256`	`1268`	`MatchingInfo info;`
`1257`	`1269`	`info.status = REMOVED_MATCHING;`
`1258`	`1270`	`info.remoteEndpointGuid = writer_guid;`
	`1271`	`+ std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());`
`1259`	`1272`	`r.get_listener()->on_reader_matched(&r, info);`
`1260`	`1273`	`}`
`1261`	`1274`	`}`
`@@ -1323,6 +1336,7 @@ bool EDP::pairing_writer_proxy_with_local_reader(`
`1323`	`1336`	`MatchingInfo info;`
`1324`	`1337`	`info.status = REMOVED_MATCHING;`
`1325`	`1338`	`info.remoteEndpointGuid = writer_guid;`
	`1339`	`+ std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());`
`1326`	`1340`	`r.get_listener()->on_reader_matched(&r, info);`
`1327`	`1341`	`}`
`1328`	`1342`	`}`
`@@ -1367,6 +1381,7 @@ bool EDP::pairing_remote_writer_with_local_reader_after_security(`
`1367`	`1381`	`MatchingInfo info;`
`1368`	`1382`	`info.status = MATCHED_MATCHING;`
`1369`	`1383`	`info.remoteEndpointGuid = writer_guid;`
	`1384`	`+ std::lock_guard<RecursiveTimedMutex> lock(r.getMutex());`
`1370`	`1385`	`r.get_listener()->on_reader_matched(&r, info);`
`1371`	`1386`
`1372`	`1387`	`}`
Original file line number	Diff line number	Diff line change
`@@ -305,7 +305,6 @@ void TypeLookupServicePublisher::on_publication_matched(`
`305`	`305`	`DataWriter* /writer/,`
`306`	`306`	`const PublicationMatchedStatus& info)`
`307`	`307`	`{`
`308`		`- std::unique_lock<std::mutex> lock(mutex_);`
`309`	`308`	`matched_ += info.current_count_change;`
`310`	`309`	`cv_.notify_all();`
`311`	`310`	`}`
Original file line number	Diff line number	Diff line change
`@@ -364,7 +364,6 @@ void TypeLookupServiceSubscriber::on_subscription_matched(`
`364`	`364`	`DataReader* /reader/,`
`365`	`365`	`const SubscriptionMatchedStatus& info)`
`366`	`366`	`{`
`367`		`- std::unique_lock<std::mutex> lock(mutex_);`
`368`	`367`	`matched_ += info.current_count_change;`
`369`	`368`	`cv_.notify_all();`
`370`	`369`	`}`