pubsub / rpc: centralize label prefiltering and matching (#299)

In SpecificDiscoveryStore, add the label set along with the handler info
to the DiscoveryCluster as "Controller" info. This allows us to do the
label matching centralized in the DiscoveryStore

* Adds a GetSupplementalDataValue function for the specified key with proper error handling included
* ServiceDiscovery: add test for two publishers and one subscriber
* Add test to test whether the optional label prefiltering works when the publishers are started before the Subscriber
* IntegrationTests: fix datarace for DataPubSub
* SpecificDiscoveryStore: improve algorithm docs
* Add comments to better explain the used algorithm, since it is somewhat convoluted

Co-authored-by: Jan Kraemer <jan.kraemer@vector.com>
Co-authored-by: Konrad Breitsprecher <Konrad.Breitsprecher@vector.com>
Co-authored-by: Marius Börschig <Marius.Boerschig@vector.com>
Co-authored-by: Daniel Edwards <Daniel.Edwards@vector.com>

Signed-off-by: Jan Kraemer <jan.kraemer@vector.com>
Signed-off-by: Konrad Breitsprecher <Konrad.Breitsprecher@vector.com>
Signed-off-by: Marius Börschig <Marius.Boerschig@vector.com>
Signed-off-by: Daniel Edwards <Daniel.Edwards@vector.com>

Author: VJanKraemer

SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp

@@ -805,4 +805,83 @@ TEST_F(ITest_Internals_DataPubSub, test_1pub_1sub_async_rejoin)
     ShutdownSystem();
 }
 
+
+// Two publishers (optional label1), one subscriber (optional label1, label2); publishers start first; subscriber joins
+TEST_F(ITest_Internals_DataPubSub, test_2pub_1sub_async_starting_order)
+{
+    const uint32_t numMsgToPublish = 1;
+    const uint32_t numMsgToReceive = 1 * numMsgToPublish;
+
+    std::vector<PubSubParticipant> publishers;
+    publishers.push_back({"Pub1",
+                          {{"PubCtrl1",
+                            "TopicA",
+                            {"A"},
+                            {{"K1", "V1", SilKit::Services::MatchingLabel::Kind::Optional}},
+                            1,
+                            defaultMsgSize,
+                            numMsgToPublish}},
+                          {}});
+    publishers.push_back({"Pub2",
+                          {{"PubCtrl1",
+                            "TopicA",
+                            {"A"},
+                            {{"K1", "V1", SilKit::Services::MatchingLabel::Kind::Optional}},
+                            1,
+                            defaultMsgSize,
+                            numMsgToPublish}},
+                          {}});
+
+    std::vector<PubSubParticipant> subscribers;
+    std::vector<std::vector<uint8_t>> expectedDataUnordered;
+    expectedDataUnordered.reserve(numMsgToReceive);
+    for (uint32_t d = 0; d < numMsgToReceive; d++)
+    {
+        // Receive the same blob several times (once from every publisher)
+        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, 0));
+    }
+    subscribers.push_back(
+        {"Sub1",
+         {},
+         {{
+             "SubCtrl1",
+             "TopicA",
+             {"A"},
+             {{"K1", "V1", SilKit::Services::MatchingLabel::Kind::Optional},
+              {"K2", "V2", SilKit::Services::MatchingLabel::Kind::Optional}
+              }, // BUGHUNT: Second label breaks communication 
+             defaultMsgSize,
+             numMsgToReceive,
+             1,
+             expectedDataUnordered,
+         }}});
+
+    for (auto& sub : subscribers)
+    {
+        sub.communicationTimeout = std::chrono::milliseconds(1000);
+    }
+
+    _testSystem.SetupRegistryAndSystemMaster("silkit://localhost:0", false, {});
+
+
+    //BUGHUNT: Subscribers start first fails SOMETIMES
+    //RunParticipants(subscribers, _testSystem.GetRegistryUri(), false);
+
+    //BUGHUNT: Publishers start first fails ALWAYS
+
+    // Start publishers
+    RunParticipants(publishers, _testSystem.GetRegistryUri(), false);
+    for (auto& p : publishers)
+    {
+        p.WaitForAllSent();
+    }
+
+    // Start subscriber
+    RunParticipants(subscribers, _testSystem.GetRegistryUri(), false);
+    
+
+    JoinPubSubThreads();
+    ShutdownSystem();
+}
+
 } // anonymous namespace

SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp

@@ -178,6 +178,7 @@ class ITest_Internals_DataPubSub : public testing::Test
             , name{newName}
             , dataSubscribers{newDataSubscribers}
             , dataPublishers{newDataPublishers}
+            , allReceived{std::make_unique<std::atomic<bool>>(false)}
         {
         }
 
@@ -186,6 +187,7 @@ class ITest_Internals_DataPubSub : public testing::Test
         std::string name;
         std::vector<DataSubscriberInfo> dataSubscribers;
         std::vector<DataPublisherInfo> dataPublishers;
+        std::unique_ptr<std::atomic<bool>> allReceived;
         std::unique_ptr<SilKit::IParticipant> participant;
         SilKit::Core::IParticipantInternal* participantImpl = nullptr;
 
@@ -196,7 +198,6 @@ class ITest_Internals_DataPubSub : public testing::Test
         std::promise<void> allDiscoveredPromise;
         bool allDiscovered{false};
         std::promise<void> allReceivedPromise;
-        bool allReceived{false};
         // Pub
         std::promise<void> allSentPromise;
         bool allSent{false};
@@ -208,7 +209,7 @@ class ITest_Internals_DataPubSub : public testing::Test
             if (std::all_of(dataSubscribers.begin(), dataSubscribers.end(),
                             [](const auto& dsInfo) { return dsInfo.numMsgToReceive == 0; }))
             {
-                allReceived = true;
+                *allReceived = true;
                 allReceivedPromise.set_value();
             }
         }
@@ -224,11 +225,11 @@ class ITest_Internals_DataPubSub : public testing::Test
 
         void CheckAllReceivedPromise()
         {
-            if (!allReceived && std::all_of(dataSubscribers.begin(), dataSubscribers.end(), [](const auto& dsInfo) {
+            if (!*allReceived && std::all_of(dataSubscribers.begin(), dataSubscribers.end(), [](const auto& dsInfo) {
                 return dsInfo.allReceived;
             }))
             {
-                allReceived = true;
+                *allReceived = true;
                 allReceivedPromise.set_value();
             }
         }

SilKit/IntegrationTests/IntegrationTestInfrastructure.hpp

@@ -25,7 +25,10 @@ class TestInfrastructure
     {
         std::stringstream ss;
         ss << "Something went wrong: " << error.what() << std::endl;
-        _systemMaster.systemController->AbortSimulation();
+        if (_systemMaster.systemController)
+        {
+            _systemMaster.systemController->AbortSimulation();
+        }
         FAIL() << ss.str();
     }
 
@@ -127,7 +130,7 @@ class TestInfrastructure
     struct SystemMaster
     {
         std::unique_ptr<IParticipant> participant;
-        SilKit::Experimental::Services::Orchestration::ISystemController* systemController;
+        SilKit::Experimental::Services::Orchestration::ISystemController* systemController{nullptr};
         ISystemMonitor* systemMonitor;
         ILifecycleService* lifecycleService;
 

SilKit/include/silkit/services/datatypes.hpp

@@ -6,6 +6,7 @@
 
 #include <cstdint>
 #include <string>
+#include <tuple>
 
 #include "silkit/util/HandlerId.hpp"
 
@@ -44,6 +45,11 @@ struct MatchingLabel
     std::string key;   //!< The label's key.
     std::string value; //!< The label's key.
     Kind kind;         //!< The matching kind to apply for this label.
+
+    friend bool operator==(const MatchingLabel& lhs, const MatchingLabel& rhs) noexcept
+    {
+        return std::tie(lhs.key, lhs.value, lhs.kind) == std::tie(rhs.key, rhs.value, rhs.kind);
+    }
 };
 
 using SilKit::Util::HandlerId;

SilKit/source/core/internal/ServiceDescriptor.hpp

@@ -86,6 +86,8 @@ class ServiceDescriptor
     inline bool GetSupplementalDataItem(const std::string& key, std::string& value) const;
     inline void SetSupplementalDataItem(std::string key, std::string val);
 
+    inline std::string GetSupplementalDataValue(const std::string& key) const;
+
     inline auto GetSimulationName() const -> const std::string&;
     inline void SetSimulationName(const std::string& simulationName);
 
@@ -137,6 +139,17 @@ void ServiceDescriptor::SetSupplementalDataItem(std::string key, std::string val
     _supplementalData[key] = std::move(val);
 }
 
+std::string ServiceDescriptor::GetSupplementalDataValue(const std::string& key) const
+{
+    std::string tmp;
+    if (GetSupplementalDataItem(key, tmp) == false)
+    {
+        throw SilKit::StateError{"Unknown key in supplementalData"};
+    }
+
+    return tmp;
+}
+
 auto ServiceDescriptor::GetParticipantId() const -> ParticipantId
 {
     return _participantId;