From 8fdcac393d34ada24117604662252e5940a729c3 Mon Sep 17 00:00:00 2001 From: Jan Kraemer Date: Wed, 4 Mar 2026 14:14:32 +0100 Subject: [PATCH 1/7] ServiceDescriptor: add generalized getVal function * Adds a getValue function for the specified key with proper error handling included Signed-off-by: Jan Kraemer --- .../source/core/internal/ServiceDescriptor.hpp | 12 ++++++++++++ .../source/services/pubsub/DataSubscriber.cpp | 16 ++++------------ SilKit/source/services/rpc/RpcClient.cpp | 10 +--------- SilKit/source/services/rpc/RpcServer.cpp | 17 ++++------------- 4 files changed, 21 insertions(+), 34 deletions(-) diff --git a/SilKit/source/core/internal/ServiceDescriptor.hpp b/SilKit/source/core/internal/ServiceDescriptor.hpp index 6b7105227..d92a001e0 100644 --- a/SilKit/source/core/internal/ServiceDescriptor.hpp +++ b/SilKit/source/core/internal/ServiceDescriptor.hpp @@ -85,6 +85,7 @@ 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 getVal(const std::string& key) const; inline auto GetSimulationName() const -> const std::string&; inline void SetSimulationName(const std::string& simulationName); @@ -137,6 +138,17 @@ void ServiceDescriptor::SetSupplementalDataItem(std::string key, std::string val _supplementalData[key] = std::move(val); } +std::string ServiceDescriptor::getVal(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; diff --git a/SilKit/source/services/pubsub/DataSubscriber.cpp b/SilKit/source/services/pubsub/DataSubscriber.cpp index 877936ab0..5342a46e0 100644 --- a/SilKit/source/services/pubsub/DataSubscriber.cpp +++ b/SilKit/source/services/pubsub/DataSubscriber.cpp @@ -31,16 +31,8 @@ void DataSubscriber::RegisterServiceDiscovery() { auto matchHandler = [this](SilKit::Core::Discovery::ServiceDiscoveryEvent::Type discoveryType, const SilKit::Core::ServiceDescriptor& serviceDescriptor) { - auto getVal = [serviceDescriptor](const std::string& key) { - std::string tmp; - if (!serviceDescriptor.GetSupplementalDataItem(key, tmp)) - { - throw SilKitError{"Unknown key in supplementalData"}; - } - return tmp; - }; - const auto pubUUID = getVal(Core::Discovery::supplKeyDataPublisherPubUUID); + const auto pubUUID = serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherPubUUID); // Early abort creation if Publisher is already connected if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceCreated @@ -49,13 +41,13 @@ void DataSubscriber::RegisterServiceDiscovery() return; } - const auto topic = getVal(Core::Discovery::supplKeyDataPublisherTopic); + const auto topic = serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherTopic); if (topic == _topic) { - const std::string pubMediaType{getVal(Core::Discovery::supplKeyDataPublisherMediaType)}; + const std::string pubMediaType{serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherMediaType)}; if (MatchMediaType(_mediaType, pubMediaType)) { - const std::string labelsStr = getVal(Core::Discovery::supplKeyDataPublisherPubLabels); + const std::string labelsStr = serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherPubLabels); const std::vector publisherLabels = SilKit::Config::Deserialize>(labelsStr); if (Util::MatchLabels(_labels, publisherLabels)) diff --git a/SilKit/source/services/rpc/RpcClient.cpp b/SilKit/source/services/rpc/RpcClient.cpp index 88df9e425..3b539ec3e 100644 --- a/SilKit/source/services/rpc/RpcClient.cpp +++ b/SilKit/source/services/rpc/RpcClient.cpp @@ -55,16 +55,8 @@ void RpcClient::RegisterServiceDiscovery() { auto matchHandler = [this](SilKit::Core::Discovery::ServiceDiscoveryEvent::Type discoveryType, const SilKit::Core::ServiceDescriptor& serviceDescriptor) { - auto getVal = [serviceDescriptor](const std::string& key) { - std::string tmp; - if (!serviceDescriptor.GetSupplementalDataItem(key, tmp)) - { - throw SilKit::StateError{"Unknown key in supplementalData"}; - } - return tmp; - }; - auto clientUUID = getVal(Core::Discovery::supplKeyRpcServerInternalClientUUID); + auto clientUUID = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcServerInternalClientUUID); if (clientUUID == _clientUUID) { diff --git a/SilKit/source/services/rpc/RpcServer.cpp b/SilKit/source/services/rpc/RpcServer.cpp index 3840d876e..2b436f714 100644 --- a/SilKit/source/services/rpc/RpcServer.cpp +++ b/SilKit/source/services/rpc/RpcServer.cpp @@ -32,16 +32,7 @@ void RpcServer::RegisterServiceDiscovery() const SilKit::Core::ServiceDescriptor& serviceDescriptor) { if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceCreated) { - auto getVal = [serviceDescriptor](const std::string& key) { - std::string tmp; - if (!serviceDescriptor.GetSupplementalDataItem(key, tmp)) - { - throw SilKit::StateError{"Unknown key in supplementalData"}; - } - return tmp; - }; - - auto clientUUID = getVal(Core::Discovery::supplKeyRpcClientUUID); + auto clientUUID = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcClientUUID); // Early abort creation if Client is already connected if (_internalRpcServers.count(clientUUID) > 0) @@ -49,9 +40,9 @@ void RpcServer::RegisterServiceDiscovery() return; } - auto functionName = getVal(Core::Discovery::supplKeyRpcClientFunctionName); - auto clientMediaType = getVal(Core::Discovery::supplKeyRpcClientMediaType); - std::string labelsStr = getVal(Core::Discovery::supplKeyRpcClientLabels); + auto functionName = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcClientFunctionName); + auto clientMediaType = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcClientMediaType); + std::string labelsStr = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcClientLabels); auto clientLabels = SilKit::Config::Deserialize>(labelsStr); if (functionName == _dataSpec.FunctionName() && MatchMediaType(clientMediaType, _dataSpec.MediaType()) From b7e98f8e7f298729cefd7b6544dba53102c322da Mon Sep 17 00:00:00 2001 From: Jan Kraemer Date: Wed, 4 Mar 2026 14:16:35 +0100 Subject: [PATCH 2/7] service: Fix condition for label prefiltering Signed-off-by: Jan Kraemer --- SilKit/source/core/service/SpecificDiscoveryStore.cpp | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/SilKit/source/core/service/SpecificDiscoveryStore.cpp b/SilKit/source/core/service/SpecificDiscoveryStore.cpp index ed7a76bcd..4cd2344dd 100644 --- a/SilKit/source/core/service/SpecificDiscoveryStore.cpp +++ b/SilKit/source/core/service/SpecificDiscoveryStore.cpp @@ -77,7 +77,7 @@ void SpecificDiscoveryStore::CallHandlerOnHandlerRegistration( const ServiceDiscoveryHandler& handler, const std::string& controllerType_, const std::string& key, const std::vector& labels) { - // pre filter key and mediaType + // pre filter controllerType and Topic/Function auto& entry = _lookup[MakeFilter(controllerType_, key)]; auto* greedyLabel = GetLabelWithMinimalNodeSet(entry, labels); @@ -211,7 +211,7 @@ auto SpecificDiscoveryStore::GetLabelWithMinimalHandlerSet(DiscoveryKeyNode& key auto& not_label_handlers = keyNode.notLabelMap[l.key].handlers; size_t relevantNodeCount = fit_handlers.size() + not_label_handlers.size(); - if (relevantNodeCount < matchCount) + if ( relevantNodeCount > 0 && relevantNodeCount < matchCount) { matchCount = relevantNodeCount; outGreedyLabel = &l; @@ -251,7 +251,7 @@ auto SpecificDiscoveryStore::GetLabelWithMinimalNodeSet(DiscoveryKeyNode& keyNod auto& not_label_nodes = keyNode.notLabelMap[l.key].nodes; size_t relevantNodeCount = fit_nodes.size() + not_label_nodes.size(); - if (relevantNodeCount < matchCount) + if ( relevantNodeCount > 0 && relevantNodeCount < matchCount) { matchCount = relevantNodeCount; outGreedyLabel = &l; From 80e52a99115a730ba43fc01e49502faa4443ad62 Mon Sep 17 00:00:00 2001 From: Jan Kraemer Date: Thu, 5 Mar 2026 14:07:40 +0100 Subject: [PATCH 3/7] ServiceDiscovery: filter labels in discoverystore 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 Signed-off-by: Jan Kraemer --- SilKit/include/silkit/services/datatypes.hpp | 6 ++ .../core/service/SpecificDiscoveryStore.cpp | 102 +++++++++++++----- .../core/service/SpecificDiscoveryStore.hpp | 17 ++- .../service/Test_SpecificDiscoveryStore.cpp | 3 +- .../source/services/pubsub/DataSubscriber.cpp | 30 +++--- SilKit/source/services/rpc/RpcServer.cpp | 4 +- SilKit/source/util/LabelMatching.cpp | 2 + 7 files changed, 116 insertions(+), 48 deletions(-) diff --git a/SilKit/include/silkit/services/datatypes.hpp b/SilKit/include/silkit/services/datatypes.hpp index ec127decb..366bcb323 100644 --- a/SilKit/include/silkit/services/datatypes.hpp +++ b/SilKit/include/silkit/services/datatypes.hpp @@ -6,6 +6,7 @@ #include #include +#include #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; diff --git a/SilKit/source/core/service/SpecificDiscoveryStore.cpp b/SilKit/source/core/service/SpecificDiscoveryStore.cpp index 4cd2344dd..605e3bf01 100644 --- a/SilKit/source/core/service/SpecificDiscoveryStore.cpp +++ b/SilKit/source/core/service/SpecificDiscoveryStore.cpp @@ -3,7 +3,9 @@ // SPDX-License-Identifier: MIT #include "SpecificDiscoveryStore.hpp" +#include "LabelMatching.hpp" #include "YamlParser.hpp" + namespace { inline auto MakeFilter(const std::string& type, const std::string& topicOrFunction) -> SilKit::Core::Discovery::FilterType @@ -87,27 +89,44 @@ void SpecificDiscoveryStore::CallHandlerOnHandlerRegistration( // no labels present trigger all for (auto&& serviceDescriptor : entry.allCluster.nodes) { - handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + const auto descriptorLabels = GetLabels(serviceDescriptor); + if(Util::MatchLabels(labels, descriptorLabels)) + { + handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + } } } else { + if (greedyLabel->kind == SilKit::Services::MatchingLabel::Kind::Optional) { // trigger notlabel handlers for (auto&& serviceDescriptor : entry.notLabelMap[greedyLabel->key].nodes) { - handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + const auto descriptorLabels = GetLabels(serviceDescriptor); + if(Util::MatchLabels(labels, descriptorLabels)) + { + handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + } } for (auto&& serviceDescriptor : entry.noLabelCluster.nodes) { - handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + const auto descriptorLabels = GetLabels(serviceDescriptor); + if(Util::MatchLabels(labels, descriptorLabels)) + { + handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + } } } // trigger label handlers for (auto&& serviceDescriptor : entry.labelMap[MakeFilter(greedyLabel->key, greedyLabel->value)].nodes) { - handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + const auto descriptorLabels = GetLabels(serviceDescriptor); + if(Util::MatchLabels(labels, descriptorLabels)) + { + handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); + } } } } @@ -126,12 +145,15 @@ void SpecificDiscoveryStore::CallHandlersOnServiceChange(ServiceDiscoveryEvent:: if (greedyLabel == nullptr) { + + bool skipLabelCheck = supplControllerTypeName == controllerTypeRpcServerInternal; // no labels present trigger all - for (auto&& handler : entry.allCluster.handlers) + for (auto&& controllerInfo : entry.allCluster.controllerInfo) { - if (handler) + bool run_handler = skipLabelCheck ? true : Util::MatchLabels(controllerInfo->labels, labels); + if (controllerInfo->handler && run_handler) { - (*handler)(eventType, serviceDescriptor); + controllerInfo->handler(eventType, serviceDescriptor); } } } @@ -140,27 +162,27 @@ void SpecificDiscoveryStore::CallHandlersOnServiceChange(ServiceDiscoveryEvent:: if (greedyLabel->kind == SilKit::Services::MatchingLabel::Kind::Optional) { // trigger notlabel handlers - for (auto&& handler : entry.notLabelMap[greedyLabel->key].handlers) + for (auto&& controllerInfo : entry.notLabelMap[greedyLabel->key].controllerInfo) { - if (handler) + if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) { - (*handler)(eventType, serviceDescriptor); + controllerInfo->handler(eventType, serviceDescriptor); } } - for (auto&& handler : entry.noLabelCluster.handlers) + for (auto&& controllerInfo : entry.noLabelCluster.controllerInfo) { - if (handler) + if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) { - (*handler)(eventType, serviceDescriptor); + controllerInfo->handler(eventType, serviceDescriptor); } } } // trigger label handlers - for (auto&& handler : entry.labelMap[MakeFilter(greedyLabel->key, greedyLabel->value)].handlers) + for (auto&& controllerInfo : entry.labelMap[MakeFilter(greedyLabel->key, greedyLabel->value)].controllerInfo) { - if (handler) + if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) { - (*handler)(eventType, serviceDescriptor); + controllerInfo->handler(eventType, serviceDescriptor); } } } @@ -188,7 +210,7 @@ auto SpecificDiscoveryStore::GetLabelWithMinimalHandlerSet(DiscoveryKeyNode& key { const SilKit::Services::MatchingLabel* outGreedyLabel = nullptr; - size_t matchCount = keyNode.allCluster.handlers.size(); + size_t matchCount = keyNode.allCluster.controllerInfo.size(); // search greedy Cluster guess for (auto&& l : labels) { @@ -197,8 +219,7 @@ auto SpecificDiscoveryStore::GetLabelWithMinimalHandlerSet(DiscoveryKeyNode& key const auto keyTuple = std::make_tuple(l.key, l.value); if (l.kind == SilKit::Services::MatchingLabel::Kind::Mandatory) { - auto& handlers = keyNode.labelMap[keyTuple].handlers; - const auto relevantNodeCount = handlers.size(); + const auto& relevantNodeCount = keyNode.labelMap[keyTuple].controllerInfo.size(); if (relevantNodeCount < matchCount) { matchCount = relevantNodeCount; @@ -207,10 +228,10 @@ auto SpecificDiscoveryStore::GetLabelWithMinimalHandlerSet(DiscoveryKeyNode& key } else if (l.kind == SilKit::Services::MatchingLabel::Kind::Optional) { - auto& fit_handlers = keyNode.labelMap[keyTuple].handlers; - auto& not_label_handlers = keyNode.notLabelMap[l.key].handlers; + const auto labeled_matches = keyNode.labelMap[keyTuple].controllerInfo.size(); + const auto distinct_matches = keyNode.notLabelMap[l.key].controllerInfo.size(); - size_t relevantNodeCount = fit_handlers.size() + not_label_handlers.size(); + const size_t relevantNodeCount = labeled_matches + distinct_matches; if ( relevantNodeCount > 0 && relevantNodeCount < matchCount) { matchCount = relevantNodeCount; @@ -292,9 +313,9 @@ void SpecificDiscoveryStore::UpdateDiscoveryClusters(const std::string& controll entry.notLabelMap[l.key].nodes.emplace_back(serviceDescriptor); } // label is seen for the first time (add all earlier handlers to notLabelEntry - for (auto& handler : entry.allCluster.handlers) + for (auto& controllerInfo : entry.allCluster.controllerInfo) { - entry.notLabelMap[l.key].handlers.emplace_back(handler); + entry.notLabelMap[l.key].controllerInfo.emplace_back(controllerInfo); } } } @@ -356,9 +377,38 @@ void SpecificDiscoveryStore::InsertLookupHandler(const std::string& controllerTy const std::vector& labels, ServiceDiscoveryHandler handler) { - auto handlerPtr = std::make_shared(std::move(handler)); + auto controllerInfo = std::make_shared(ControllerCluster(std::move(handler), labels)); UpdateDiscoveryClusters(controllerType_, key, labels, - [handlerPtr](auto& cluster) { cluster.handlers.push_back(handlerPtr); }); + [controllerInfo](auto& cluster) { cluster.controllerInfo.push_back(controllerInfo); }); +} + +const std::vector SpecificDiscoveryStore::GetLabels( + const ServiceDescriptor& descriptor) +{ + + + const auto ctrlType = descriptor.getVal(Core::Discovery::controllerType); + + std::string labelsStr; + + if(ctrlType == controllerTypeDataPublisher) + { + labelsStr = descriptor.getVal(Core::Discovery::supplKeyDataPublisherPubLabels); + } + else if(ctrlType == controllerTypeRpcClient) + { + labelsStr = descriptor.getVal(Core::Discovery::supplKeyRpcClientLabels); + } + else + { + // Don't need labels return an empty vector + return std::vector(); + } + + const auto descriptorLabels = + SilKit::Config::Deserialize>(labelsStr); + return descriptorLabels; + } void SpecificDiscoveryStore::RegisterSpecificServiceDiscoveryHandler( diff --git a/SilKit/source/core/service/SpecificDiscoveryStore.hpp b/SilKit/source/core/service/SpecificDiscoveryStore.hpp index b3ed137f2..3c095db34 100644 --- a/SilKit/source/core/service/SpecificDiscoveryStore.hpp +++ b/SilKit/source/core/service/SpecificDiscoveryStore.hpp @@ -10,6 +10,7 @@ #include #include #include +#include #include "IServiceDiscovery.hpp" #include "Hash.hpp" @@ -33,12 +34,23 @@ struct FilterTypeHash using HandlerValue = std::shared_ptr; +struct ControllerCluster { + ServiceDiscoveryHandler handler; + std::vector labels; + + ControllerCluster(ServiceDiscoveryHandler ahandler, const std::vector& alabels) : + handler(ahandler), + labels(alabels) { + }; + +}; + //! Stores all potential nodes (service descriptors) and handlers to call for a specific data matching branch class DiscoveryCluster { public: std::vector nodes; - std::vector handlers; + std::vector> controllerInfo; }; //! Holds all relevant information for a controllerType and key (topic/functionName/clientUUID) @@ -126,6 +138,9 @@ class SpecificDiscoveryStore const std::vector& labels, ServiceDiscoveryHandler handler); + //!< Get serviceDescriptorLabels + const std::vector GetLabels(const ServiceDescriptor& descriptor); + private: //member //!< SpecificDiscoveryStore is only available to a a sub set of controllers const std::unordered_set _allowedControllers = { diff --git a/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp b/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp index c567d864d..eabdab416 100644 --- a/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp +++ b/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp @@ -4,7 +4,6 @@ #include #include -#include #include #include "gtest/gtest.h" @@ -310,7 +309,7 @@ TEST_F(Test_SpecificDiscoveryStore, lookup_service_discovery_then_handler_labels testStore.ServiceChange(ServiceDiscoveryEvent::Type::ServiceCreated, noLabelTestDescriptor); EXPECT_CALL(callbacks, ServiceDiscoveryHandler(ServiceDiscoveryEvent::Type::ServiceCreated, noLabelTestDescriptor)) - .Times(1); + .Times(0); testStore.RegisterSpecificServiceDiscoveryHandler( [this](ServiceDiscoveryEvent::Type discoveryType, const ServiceDescriptor& sd) { diff --git a/SilKit/source/services/pubsub/DataSubscriber.cpp b/SilKit/source/services/pubsub/DataSubscriber.cpp index 5342a46e0..c1a293638 100644 --- a/SilKit/source/services/pubsub/DataSubscriber.cpp +++ b/SilKit/source/services/pubsub/DataSubscriber.cpp @@ -42,27 +42,23 @@ void DataSubscriber::RegisterServiceDiscovery() } const auto topic = serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherTopic); - if (topic == _topic) + + // We need to just match the MediaType, the topic and labels were already prefiltered by the ServiceDiscovery + const std::string pubMediaType{serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherMediaType)}; + if (MatchMediaType(_mediaType, pubMediaType)) { - const std::string pubMediaType{serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherMediaType)}; - if (MatchMediaType(_mediaType, pubMediaType)) + std::unique_lock lock(_internalSubscribersMx); + + if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceCreated) { const std::string labelsStr = serviceDescriptor.getVal(Core::Discovery::supplKeyDataPublisherPubLabels); - const std::vector publisherLabels = + const auto publisherLabels = SilKit::Config::Deserialize>(labelsStr); - if (Util::MatchLabels(_labels, publisherLabels)) - { - std::unique_lock lock(_internalSubscribersMx); - - if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceCreated) - { - AddInternalSubscriber(pubUUID, pubMediaType, publisherLabels); - } - else if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceRemoved) - { - RemoveInternalSubscriber(pubUUID); - } - } + AddInternalSubscriber(pubUUID, pubMediaType, publisherLabels); + } + else if (discoveryType == SilKit::Core::Discovery::ServiceDiscoveryEvent::Type::ServiceRemoved) + { + RemoveInternalSubscriber(pubUUID); } } }; diff --git a/SilKit/source/services/rpc/RpcServer.cpp b/SilKit/source/services/rpc/RpcServer.cpp index 2b436f714..5133fc3c8 100644 --- a/SilKit/source/services/rpc/RpcServer.cpp +++ b/SilKit/source/services/rpc/RpcServer.cpp @@ -45,8 +45,8 @@ void RpcServer::RegisterServiceDiscovery() std::string labelsStr = serviceDescriptor.getVal(Core::Discovery::supplKeyRpcClientLabels); auto clientLabels = SilKit::Config::Deserialize>(labelsStr); - if (functionName == _dataSpec.FunctionName() && MatchMediaType(clientMediaType, _dataSpec.MediaType()) - && Util::MatchLabels(_dataSpec.Labels(), clientLabels)) + // Match only on the MediaType, FunctionName and Labels are already prefiltered by the DiscoveryService + if (MatchMediaType(clientMediaType, _dataSpec.MediaType())) { AddInternalRpcServer(clientUUID, clientMediaType, clientLabels); } diff --git a/SilKit/source/util/LabelMatching.cpp b/SilKit/source/util/LabelMatching.cpp index 671e58137..e70cd6d00 100644 --- a/SilKit/source/util/LabelMatching.cpp +++ b/SilKit/source/util/LabelMatching.cpp @@ -4,6 +4,7 @@ #include "LabelMatching.hpp" #include +#include namespace SilKit { namespace Util { @@ -62,6 +63,7 @@ bool MatchLabels(const std::vector& labels1, const std::vector match } From 1a95a534a0fee586ac8c52c85cd6487617243e91 Mon Sep 17 00:00:00 2001 From: Konrad Breitsprecher Date: Thu, 29 Jan 2026 09:48:06 +0100 Subject: [PATCH 4/7] ServiceDiscovery: add test 2pubs one sub * Add test to test whether the optional label prefiltering works when the publishers are started before the Subscriber Signed-off-by: Konrad Breitsprecher Co-authored-by: Jan Kraemer --- .../ITest_Internals_DataPubSub.cpp | 79 +++++++++++++++++++ .../ITest_Internals_DataPubSub.hpp | 2 +- .../IntegrationTestInfrastructure.hpp | 7 +- 3 files changed, 85 insertions(+), 3 deletions(-) diff --git a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp index 641838918..16408053e 100644 --- a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp +++ b/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 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 subscribers; + std::vector> 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(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 diff --git a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp index 107cc9efe..e6778dbb0 100644 --- a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp +++ b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp @@ -380,7 +380,7 @@ class ITest_Internals_DataPubSub : public testing::Test participant.allSentPromise.set_value(); } - if (!participant.dataSubscribers.empty()) + if (!participant.dataSubscribers.empty() && !participant.allReceived) { participant.WaitForAllReceived(); } diff --git a/SilKit/IntegrationTests/IntegrationTestInfrastructure.hpp b/SilKit/IntegrationTests/IntegrationTestInfrastructure.hpp index d38f77ae2..1a2d12135 100644 --- a/SilKit/IntegrationTests/IntegrationTestInfrastructure.hpp +++ b/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 participant; - SilKit::Experimental::Services::Orchestration::ISystemController* systemController; + SilKit::Experimental::Services::Orchestration::ISystemController* systemController{nullptr}; ISystemMonitor* systemMonitor; ILifecycleService* lifecycleService; From 87efda661071d3532e1bf6b50dbac959234391fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Wed, 11 Feb 2026 12:22:27 +0100 Subject: [PATCH 5/7] IntegrationTests: fix datarace for DataPubSub MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marius Börschig Co-authored-by: Jan Kraemer --- .../IntegrationTests/ITest_Internals_DataPubSub.hpp | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp index e6778dbb0..1ff4f15ee 100644 --- a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp +++ b/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>(false)} { } @@ -186,6 +187,7 @@ class ITest_Internals_DataPubSub : public testing::Test std::string name; std::vector dataSubscribers; std::vector dataPublishers; + std::unique_ptr> allReceived; std::unique_ptr participant; SilKit::Core::IParticipantInternal* participantImpl = nullptr; @@ -196,7 +198,6 @@ class ITest_Internals_DataPubSub : public testing::Test std::promise allDiscoveredPromise; bool allDiscovered{false}; std::promise allReceivedPromise; - bool allReceived{false}; // Pub std::promise 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(); } } @@ -380,7 +381,7 @@ class ITest_Internals_DataPubSub : public testing::Test participant.allSentPromise.set_value(); } - if (!participant.dataSubscribers.empty() && !participant.allReceived) + if (!participant.dataSubscribers.empty()) { participant.WaitForAllReceived(); } From 300dc498ad8fbde09fd917ecdca75a28875fd46b Mon Sep 17 00:00:00 2001 From: Jan Kraemer Date: Thu, 12 Mar 2026 10:04:28 +0100 Subject: [PATCH 6/7] SpecificDiscoveryStore: improve algorithm docs * Add comments to better explain the used algorithm, since it is somewhat convoluted Signed-off-by: Jan Kraemer --- .../source/core/service/SpecificDiscoveryStore.cpp | 10 ++++++---- .../source/core/service/SpecificDiscoveryStore.hpp | 14 ++++++++++++-- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/SilKit/source/core/service/SpecificDiscoveryStore.cpp b/SilKit/source/core/service/SpecificDiscoveryStore.cpp index 605e3bf01..c5111c908 100644 --- a/SilKit/source/core/service/SpecificDiscoveryStore.cpp +++ b/SilKit/source/core/service/SpecificDiscoveryStore.cpp @@ -101,7 +101,7 @@ void SpecificDiscoveryStore::CallHandlerOnHandlerRegistration( if (greedyLabel->kind == SilKit::Services::MatchingLabel::Kind::Optional) { - // trigger notlabel handlers + // Get all services that do not have the same optional label present for (auto&& serviceDescriptor : entry.notLabelMap[greedyLabel->key].nodes) { const auto descriptorLabels = GetLabels(serviceDescriptor); @@ -110,6 +110,7 @@ void SpecificDiscoveryStore::CallHandlerOnHandlerRegistration( handler(ServiceDiscoveryEvent::Type::ServiceCreated, serviceDescriptor); } } + // Get all services that do not have any labels attached, thus matching our optional label for (auto&& serviceDescriptor : entry.noLabelCluster.nodes) { const auto descriptorLabels = GetLabels(serviceDescriptor); @@ -119,7 +120,7 @@ void SpecificDiscoveryStore::CallHandlerOnHandlerRegistration( } } } - // trigger label handlers + // trigger label handlers for exact matches (optional and mandatory) for (auto&& serviceDescriptor : entry.labelMap[MakeFilter(greedyLabel->key, greedyLabel->value)].nodes) { const auto descriptorLabels = GetLabels(serviceDescriptor); @@ -161,7 +162,7 @@ void SpecificDiscoveryStore::CallHandlersOnServiceChange(ServiceDiscoveryEvent:: { if (greedyLabel->kind == SilKit::Services::MatchingLabel::Kind::Optional) { - // trigger notlabel handlers + // trigger handlers that do not have the same optional label for (auto&& controllerInfo : entry.notLabelMap[greedyLabel->key].controllerInfo) { if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) @@ -169,6 +170,7 @@ void SpecificDiscoveryStore::CallHandlersOnServiceChange(ServiceDiscoveryEvent:: controllerInfo->handler(eventType, serviceDescriptor); } } + // trigger handlers with no labels attached, thus matching our optional label for (auto&& controllerInfo : entry.noLabelCluster.controllerInfo) { if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) @@ -177,7 +179,7 @@ void SpecificDiscoveryStore::CallHandlersOnServiceChange(ServiceDiscoveryEvent:: } } } - // trigger label handlers + // trigger label handlers with exact matches (optional and mandatory) for (auto&& controllerInfo : entry.labelMap[MakeFilter(greedyLabel->key, greedyLabel->value)].controllerInfo) { if (controllerInfo->handler && Util::MatchLabels(controllerInfo->labels, labels)) diff --git a/SilKit/source/core/service/SpecificDiscoveryStore.hpp b/SilKit/source/core/service/SpecificDiscoveryStore.hpp index 3c095db34..b317d980d 100644 --- a/SilKit/source/core/service/SpecificDiscoveryStore.hpp +++ b/SilKit/source/core/service/SpecificDiscoveryStore.hpp @@ -114,12 +114,22 @@ class SpecificDiscoveryStore const std::vector& labels, std::function); - //!< Looks for the label that returns a minimal handler set + /*! \brief Looks for the label that returns a minimal handler set + * + * For a suitable controller, ALL its labels must match ALL of our controllers labels + * So we can preselect them via checking which of the controllers labels matches with the least amount of services + * since the others have labels present that are not in the controllers label list, thus they are never going to match. + */ auto GetLabelWithMinimalHandlerSet(DiscoveryKeyNode& keyNode, const std::vector& labels) -> const SilKit::Services::MatchingLabel*; - //!< Looks for the label that returns a minimal ServiceDescriptor set + /*! \brief Looks for the label that returns a minimal ServiceDescriptor set + * + * For a suitable service, ALL its labels must match ALL of our controllers labels + * So we can preselect them via checking which of the stored service labels match with the least amount of handlers + * since the others have labels present that are not in the services label list, they are never going to match. + */ auto GetLabelWithMinimalNodeSet(DiscoveryKeyNode& keyNode, const std::vector& labels) -> const SilKit::Services::MatchingLabel*; From ac017cbd3334e61a55167bab140900801c3af127 Mon Sep 17 00:00:00 2001 From: Jan Kraemer Date: Thu, 12 Mar 2026 10:05:17 +0100 Subject: [PATCH 7/7] SpecificDiscoveryStore: fix variable names in test Signed-off-by: Jan Kraemer --- .../source/core/service/Test_SpecificDiscoveryStore.cpp | 8 ++++---- SilKit/source/util/LabelMatching.cpp | 1 - 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp b/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp index eabdab416..9eec69aae 100644 --- a/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp +++ b/SilKit/source/core/service/Test_SpecificDiscoveryStore.cpp @@ -303,12 +303,12 @@ TEST_F(Test_SpecificDiscoveryStore, lookup_service_discovery_then_handler_labels baseDescriptor.SetSupplementalDataItem(supplKeyDataPublisherMediaType, "text/json"); baseDescriptor.SetSupplementalDataItem(supplKeyDataPublisherPubLabels, "- key: kA\n value: vA\n kind: 2"); - ServiceDescriptor noLabelTestDescriptor{baseDescriptor}; - noLabelTestDescriptor.SetServiceId(1); + ServiceDescriptor testDescriptor{baseDescriptor}; + testDescriptor.SetServiceId(1); - testStore.ServiceChange(ServiceDiscoveryEvent::Type::ServiceCreated, noLabelTestDescriptor); + testStore.ServiceChange(ServiceDiscoveryEvent::Type::ServiceCreated, testDescriptor); - EXPECT_CALL(callbacks, ServiceDiscoveryHandler(ServiceDiscoveryEvent::Type::ServiceCreated, noLabelTestDescriptor)) + EXPECT_CALL(callbacks, ServiceDiscoveryHandler(ServiceDiscoveryEvent::Type::ServiceCreated, testDescriptor)) .Times(0); testStore.RegisterSpecificServiceDiscoveryHandler( diff --git a/SilKit/source/util/LabelMatching.cpp b/SilKit/source/util/LabelMatching.cpp index e70cd6d00..6e2b969ac 100644 --- a/SilKit/source/util/LabelMatching.cpp +++ b/SilKit/source/util/LabelMatching.cpp @@ -4,7 +4,6 @@ #include "LabelMatching.hpp" #include -#include namespace SilKit { namespace Util {