Refs #21355. Added regression test.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

Author: MiguelCompany

test/blackbox/common/DDSBlackboxTestsDiscovery.cpp

@@ -483,6 +483,165 @@ TEST(DDSDiscovery, UpdateMatchedStatus)
     datawriter_2.destroy();
 }
 
+/*
+ * This is a regression test for redmine issue 21355.
+ *
+ * In order to check that the on_publication_matched() and on_subscription_matched() callbacks are always called with
+ * either +1 or -1 as the current_count_change, this test creates and destroys multiple DataReaders and DataWriters in
+ * different threads, with a listener that counts the number of times on_publication_matched() and
+ * on_subscription_matched() are called with current_count_change equal to 1, -1 or other value.
+ *
+ * The test fails if any of the calls to on_publication_matched() or on_subscription_matched() is made with a
+ * current_count_change different than 1 or -1, or if the total number of calls with current_count_change equal to 1
+ * is different than the total number of calls with current_count_change equal to -1.
+ */
+TEST(DDSDiscovery, MatchedCallbackListenerMultithread)
+{
+    constexpr size_t NUM_THREADS = 25;
+
+    struct TestListener
+        : public eprosima::fastdds::dds::DataReaderListener
+        , public eprosima::fastdds::dds::DataWriterListener
+    {
+        std::atomic<uint32_t> reader_plus_one_calls{ 0 };
+        std::atomic<uint32_t> reader_minus_one_calls{ 0 };
+        std::atomic<uint32_t> reader_other_calls{ 0 };
+        std::atomic<uint32_t> reader_total_calls{ 0 };
+
+        std::atomic<uint32_t> writer_plus_one_calls{ 0 };
+        std::atomic<uint32_t> writer_minus_one_calls{ 0 };
+        std::atomic<uint32_t> writer_other_calls{ 0 };
+        std::atomic<uint32_t> writer_total_calls{ 0 };
+
+        void on_publication_matched(
+                eprosima::fastdds::dds::DataWriter*,
+                const eprosima::fastdds::dds::PublicationMatchedStatus& info) override
+        {
+            if (info.current_count_change == 1)
+            {
+                ++writer_plus_one_calls;
+            }
+            else if (info.current_count_change == -1)
+            {
+                ++writer_minus_one_calls;
+            }
+            else
+            {
+                ++writer_other_calls;
+            }
+
+            ++writer_total_calls;
+        }
+
+        void on_subscription_matched(
+                eprosima::fastdds::dds::DataReader*,
+                const eprosima::fastdds::dds::SubscriptionMatchedStatus& info) override
+        {
+            if (info.current_count_change == 1)
+            {
+                ++reader_plus_one_calls;
+            }
+            else if (info.current_count_change == -1)
+            {
+                ++reader_minus_one_calls;
+            }
+            else
+            {
+                ++reader_other_calls;
+            }
+
+            ++reader_total_calls;
+        }
+
+    };
+
+    std::ostringstream t;
+    t << TEST_TOPIC_NAME << "_" << asio::ip::host_name() << "_" << GET_PID();
+    std::string topic_name = t.str();
+    TestListener listener;
+    eprosima::fastdds::dds::TypeSupport type(new HelloWorldPubSubType());
+
+    auto create_entities = [&listener, &topic_name, &type]()
+            {
+                using namespace eprosima::fastdds::dds;
+
+                uint32_t domain_id = static_cast<uint32_t>(GET_PID()) % 100;
+                auto factory = DomainParticipantFactory::get_shared_instance();
+                DomainParticipantQos participant_qos;
+                factory->get_default_participant_qos(participant_qos);
+                participant_qos.setup_transports(eprosima::fastdds::rtps::BuiltinTransports::UDPv4);
+                participant_qos.wire_protocol().builtin.discovery_config.leaseDuration.seconds = 0;
+                participant_qos.wire_protocol().builtin.discovery_config.leaseDuration.nanosec = 100000000;
+                participant_qos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod.seconds = 0;
+                participant_qos.wire_protocol().builtin.discovery_config.leaseDuration_announcementperiod.nanosec =
+                        50000000;
+
+                auto participant = factory->create_participant(domain_id, participant_qos);
+                ASSERT_NE(participant, nullptr);
+
+                type.register_type(participant);
+                auto topic = participant->create_topic(topic_name, type.get_type_name(), TOPIC_QOS_DEFAULT);
+                ASSERT_NE(topic, nullptr);
+
+                auto publisher = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
+                ASSERT_NE(publisher, nullptr);
+
+                DataWriterQos datawriter_qos;
+                publisher->get_default_datawriter_qos(datawriter_qos);
+                datawriter_qos.data_sharing().off();
+                auto datawriter = publisher->create_datawriter(topic, datawriter_qos, &listener);
+                ASSERT_NE(datawriter, nullptr);
+
+                auto subscriber = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
+                ASSERT_NE(subscriber, nullptr);
+
+                DataReaderQos datareader_qos;
+                subscriber->get_default_datareader_qos(datareader_qos);
+                datareader_qos.data_sharing().off();
+                auto datareader = subscriber->create_datareader(topic, datareader_qos, &listener);
+                ASSERT_NE(datareader, nullptr);
+
+                HelloWorld msg;
+                msg.index(1);
+                datawriter->write(&msg);
+
+                SampleInfo info;
+                datareader->take_next_sample(&msg, &info);
+
+                participant->delete_contained_entities();
+                factory->delete_participant(participant);
+
+                std::this_thread::sleep_for(std::chrono::milliseconds(200));
+            };
+
+    std::vector<std::thread> threads;
+    for (int i = 0; i < NUM_THREADS; ++i)
+    {
+        threads.emplace_back(create_entities);
+    }
+
+    for (auto& thread : threads)
+    {
+        thread.join();
+    }
+
+    std::cout << "Reader +1 calls:    " << listener.reader_plus_one_calls.load() << std::endl;
+    std::cout << "Reader -1 calls:    " << listener.reader_minus_one_calls.load() << std::endl;
+    std::cout << "Reader other calls: " << listener.reader_other_calls.load() << std::endl;
+    std::cout << "Reader total calls: " << listener.reader_total_calls.load() << std::endl;
+
+    std::cout << "Writer +1 calls:    " << listener.writer_plus_one_calls.load() << std::endl;
+    std::cout << "Writer -1 calls:    " << listener.writer_minus_one_calls.load() << std::endl;
+    std::cout << "Writer other calls: " << listener.writer_other_calls.load() << std::endl;
+    std::cout << "Writer total calls: " << listener.writer_total_calls.load() << std::endl;
+
+
+    EXPECT_EQ(listener.reader_other_calls.load(), 0u);
+    EXPECT_EQ(listener.writer_other_calls.load(), 0u);
+    EXPECT_GE(listener.reader_plus_one_calls.load(), listener.reader_minus_one_calls.load());
+    EXPECT_GE(listener.writer_plus_one_calls.load(), listener.writer_minus_one_calls.load());
+}
+
 TEST(DDSDiscovery, EndpointMatchingCallbackAlwaysTrue)
 {
     using namespace std::chrono_literals;