diff --git a/Demos/tools/Benchmark/performance-tests.toml b/Demos/tools/Benchmark/performance-tests.toml index 42d4e3132..19eb988a0 100644 --- a/Demos/tools/Benchmark/performance-tests.toml +++ b/Demos/tools/Benchmark/performance-tests.toml @@ -80,24 +80,9 @@ args = [ ] -# NB: the tests above all run with the default of two participants, i.e. a fan-out of one, so they -# cannot observe the cost of sending a single message to many peers. The following two do. -# -# Deliveries scale with participants*(participants-1), so moving from 2 to 8 participants -# multiplies them by 28 and the wall clock with them. Shorten the simulation duration to pay -# for that, and do NOT lower the message count: the message count sets the load per simulation -# step, and the run has to stay saturated (reported speedup below 1) for the KPI to measure -# capacity rather than the virtual time step pacing. An unsaturated run is insensitive to the -# very costs these tests exist to detect. -# -# Measured on one machine, one run, 8 participants, 16 byte messages: -# message-count 200, duration 10 -> 120s, speedup 0.084, 936k msg/s -# message-count 50, duration 2 -> 6s, speedup 0.333, 933k msg/s (kept) -# message-count 10, duration 10 -> 8s, speedup 1.26, 704k msg/s (unsaturated, rejected) -# and with 65536 byte messages: -# message-count 4, duration 10 -> 19s, speedup 0.516, 7.22e3 MiB/s -# message-count 4, duration 2 -> 3s, speedup 0.664, 9.29e3 MiB/s (kept) -# message-count 2, duration 5 -> 5s, speedup 1.14, 7.99e3 MiB/s (unsaturated, rejected) +# NB: the fan-out tests use 8 participants, which multiplies deliveries by 28, hence the short +# simulation duration. Do not lower the message count: the run must stay saturated (speedup +# below 1), otherwise the KPI measures the time step pacing instead of capacity. [[tests]] name = "throughput-fanout-large-msg" unit = "MiB/s" diff --git a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp index 2fb3955e8..d74392102 100644 --- a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp +++ b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.cpp @@ -196,6 +196,46 @@ TEST_F(ITest_Internals_DataPubSub, test_1pub_2sub_sync) RunSyncTest(pubsubs); } +// One publisher participant, three subscriber participants. With more than one remote receiver the +// publisher serializes each message once and shares the body between the peers. +const std::string messageAggregationOn{R"({"Experimental": {"TimeSynchronization": {"EnableMessageAggregation": "On"}}})"}; +const std::string receiveBufferPoolOff{R"({"Experimental": {"UseReceiveBufferPool": false}})"}; + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_100B) +{ + RunFanOutSyncTest(3, 100, ""); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_4KiB) +{ + RunFanOutSyncTest(3, 4096, ""); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_largemsg) +{ + RunFanOutSyncTest(3, 250000, ""); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_100B_aggregation) +{ + RunFanOutSyncTest(3, 100, messageAggregationOn); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_4KiB_aggregation) +{ + RunFanOutSyncTest(3, 4096, messageAggregationOn); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_largemsg_aggregation) +{ + RunFanOutSyncTest(3, 250000, messageAggregationOn); +} + +TEST_F(ITest_Internals_DataPubSub, test_1pub_3sub_sync_4KiB_receive_buffer_pool_off) +{ + RunFanOutSyncTest(3, 4096, receiveBufferPoolOff); +} + // Two publisher participants, one subscriber participant on same topic: Expect all to arrive but arbitrary reception order TEST_F(ITest_Internals_DataPubSub, test_2pub_1sub_sync) { @@ -211,8 +251,8 @@ TEST_F(ITest_Internals_DataPubSub, test_2pub_1sub_sync) std::vector> expectedDataUnordered; for (uint8_t d = 0; d < numMsgToPublish; d++) { - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); } pubsubs.push_back({"Sub1", {}, @@ -246,8 +286,8 @@ TEST_F(ITest_Internals_DataPubSub, test_2pub_2sub_sync_sametopic) std::vector> expectedDataUnordered; for (uint8_t d = 0; d < numMsgToPublish; d++) { - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); } pubsubs.push_back({"Sub1", @@ -299,9 +339,9 @@ TEST_F(ITest_Internals_DataPubSub, test_3pub_4sub_sync_4topics) std::vector> expectedDataUnordered; for (uint8_t d = 0; d < numMsgToPublish; d++) { - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); } pubsubs.push_back({"Sub1", {}, @@ -698,8 +738,8 @@ TEST_F(ITest_Internals_DataPubSub, test_1_participant_selfdelivery_same_topic) std::vector> expectedDataUnordered; for (uint8_t d = 0; d < numMsgToPublish; d++) { - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); - expectedDataUnordered.emplace_back(std::vector(defaultMsgSize, d)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast(d))); } pubsubs.push_back({"PubSub1", {{"PubCtrl1", "TopicA", {"A"}, {}, 0, defaultMsgSize, numMsgToPublish}, @@ -759,7 +799,7 @@ TEST_F(ITest_Internals_DataPubSub, test_1pub_1sub_async_rejoin) 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)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, 0)); } subscribers.push_back({"Sub1", {}, @@ -838,7 +878,7 @@ TEST_F(ITest_Internals_DataPubSub, test_2pub_1sub_async_starting_order) 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)); + expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, 0)); } subscribers.push_back( {"Sub1", diff --git a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp index d5ed7fa1b..5aa8bbbed 100644 --- a/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp +++ b/SilKit/IntegrationTests/ITest_Internals_DataPubSub.hpp @@ -9,6 +9,7 @@ #include "silkit/services/all.hpp" #include "config/ConfigurationTestUtils.hpp" +#include "config/ParticipantConfigurationFromXImpl.hpp" #include "core/internal/IParticipantInternal.hpp" #include "core/service/IServiceDiscovery.hpp" #include "core/service/ServiceDatatypes.hpp" @@ -24,6 +25,17 @@ using namespace SilKit::Services::PubSub; const size_t defaultMsgSize = 3; const uint32_t defaultNumMsgToPublish = 3; +//! Non-uniform, so that shifted, truncated or mixed up payload bytes are detected. +inline auto MakeTestData(size_t size, uint8_t seed) -> std::vector +{ + std::vector data(size); + for (size_t i = 0; i < size; ++i) + { + data[i] = static_cast(i * 31 + seed); + } + return data; +} + class ITest_Internals_DataPubSub : public testing::Test { protected: @@ -60,7 +72,7 @@ class ITest_Internals_DataPubSub : public testing::Test { if (!allSent) { - auto data = std::vector(messageSizeInBytes, static_cast(publishMsgCounter)); + auto data = MakeTestData(messageSizeInBytes, static_cast(publishMsgCounter)); dataPublisher->Publish(data); publishMsgCounter++; if (publishMsgCounter >= numMsgToPublish) @@ -126,8 +138,7 @@ class ITest_Internals_DataPubSub : public testing::Test { if (expectIncreasingData) { - auto expectedData = - std::vector(messageSizeInBytes, static_cast(receiveMsgCounter)); + auto expectedData = MakeTestData(messageSizeInBytes, static_cast(receiveMsgCounter)); EXPECT_TRUE(SilKit::Util::ItemsAreEqual(dataMessageEvent.data, SilKit::Util::ToSpan(expectedData))); } else @@ -482,6 +493,34 @@ class ITest_Internals_DataPubSub : public testing::Test ShutdownSystem(); } + //! One publisher and numSubscribers subscriber participants on one topic, all using participantConfig. + void RunFanOutSyncTest(size_t numSubscribers, size_t messageSize, const std::string& participantConfig) + { + const uint32_t numMsgToPublish = defaultNumMsgToPublish; + const uint32_t numMsgToReceive = numMsgToPublish; + + auto makeConfig = [&participantConfig]() -> std::shared_ptr { + if (participantConfig.empty()) + { + return SilKit::Config::MakeEmptyParticipantConfigurationImpl(); + } + return SilKit::Config::ParticipantConfigurationFromStringImpl(participantConfig); + }; + + std::vector pubsubs; + pubsubs.push_back( + {"Pub1", {{"PubCtrl1", "TopicA", {"A"}, {}, 0, messageSize, numMsgToPublish}}, {}, makeConfig()}); + for (size_t i = 1; i <= numSubscribers; ++i) + { + pubsubs.push_back({"Sub" + std::to_string(i), + {}, + {{"SubCtrl1", "TopicA", {"A"}, {}, messageSize, numMsgToReceive, 1}}, + makeConfig()}); + } + + RunSyncTest(pubsubs); + } + void RunAsyncTest(std::vector& publishers, std::vector& subscribers) { _testSystem.SetupRegistryAndSystemMaster("silkit://localhost:0", false, {}); diff --git a/SilKit/IntegrationTests/ITest_Internals_Rpc.cpp b/SilKit/IntegrationTests/ITest_Internals_Rpc.cpp index a66c48c99..65c2ab0fb 100644 --- a/SilKit/IntegrationTests/ITest_Internals_Rpc.cpp +++ b/SilKit/IntegrationTests/ITest_Internals_Rpc.cpp @@ -179,6 +179,39 @@ TEST_F(ITest_Internals_Rpc, test_1client_2server_sync) RunSyncTest(rpcs); } +// One client participant, two server participants, with calls large enough that the client serializes +// each call once and shares the body between both servers +TEST_F(ITest_Internals_Rpc, test_1client_2server_sync_4KiB) +{ + const size_t messageSize = 4096; + const uint32_t numCallsToReceive = defaultNumCalls; + const uint32_t numCallsToReturn = defaultNumCalls * 2; + + std::vector> expectedReturnDataUnordered; + for (uint8_t d = 0; d < defaultNumCalls; d++) + { + expectedReturnDataUnordered.emplace_back(std::vector(messageSize, d + rpcFuncIncrement)); + expectedReturnDataUnordered.emplace_back(std::vector(messageSize, d + rpcFuncIncrement)); + } + + std::vector rpcs; + rpcs.push_back({"Client1", + {}, + {{"ClientCtrl1", + "TestFuncA", + "A", + {}, + messageSize, + defaultNumCalls, + numCallsToReturn, + expectedReturnDataUnordered}}, + {"TestFuncA"}}); + rpcs.push_back({"Server1", {{"ServerCtrl1", "TestFuncA", "A", {}, messageSize, numCallsToReceive}}, {}, {}}); + rpcs.push_back({"Server2", {{"ServerCtrl1", "TestFuncA", "A", {}, messageSize, numCallsToReceive}}, {}, {}}); + + RunSyncTest(rpcs); +} + // Two client participants, one server participant TEST_F(ITest_Internals_Rpc, test_Nclient_1server_sync) { diff --git a/SilKit/IntegrationTests/ITest_RequestRemoteParticipantConnect.cpp b/SilKit/IntegrationTests/ITest_RequestRemoteParticipantConnect.cpp index 33b3a097a..23d37f654 100644 --- a/SilKit/IntegrationTests/ITest_RequestRemoteParticipantConnect.cpp +++ b/SilKit/IntegrationTests/ITest_RequestRemoteParticipantConnect.cpp @@ -238,5 +238,72 @@ TEST(ITest_RequestRemoteParticipantConnect, test_fallback_to_proxy_if_registry_d (void)simTestHarness.GetParticipant("P2", tcpPCfgWithProxyAndRconn); } +const std::string anyPCfgWithProxy = R"( +Middleware: + RegistryAsFallbackProxy: true + ConnectAttempts: 1 + +Logging: + Sinks: + - Type: Stdout + #Level: Trace +)"; + +TEST(ITest_RequestRemoteParticipantConnect, test_publish_to_a_direct_and_a_proxied_subscriber) +{ + // P1 only accepts local-domain connections and P2 does not use them, so P2 reaches P1 through the + // registry proxy. P3 connects to both directly. P1 then publishes to one direct and one proxied peer, + // with payloads large enough that the message body is shared between both. + SimTestHarnessArgs simTestHarnessArgs; + simTestHarnessArgs.syncParticipantNames = {"P1", "P2", "P3"}; + simTestHarnessArgs.deferParticipantCreation = true; + simTestHarnessArgs.registry.participantConfiguration = registryConfigWithProxyNoRconn; + + SimTestHarness testSetup{simTestHarnessArgs}; + + auto* publisherParticipant = testSetup.GetParticipant("P1", locPCfgWithProxyAndRconn); + auto* proxiedParticipant = testSetup.GetParticipant("P2", tcpPCfgWithProxyAndRconn); + auto* directParticipant = testSetup.GetParticipant("P3", anyPCfgWithProxy); + + std::vector payload(4096); + for (size_t i = 0; i < payload.size(); ++i) + { + payload[i] = static_cast(i * 31 + 7); + } + + const auto pubSubSpec = SilKit::Services::PubSub::PubSubSpec{}; + uint64_t numReceivedProxied{}; + uint64_t numReceivedDirect{}; + + proxiedParticipant->Participant()->CreateDataSubscriber("test", pubSubSpec, [&](auto&&, auto&& data) { + EXPECT_EQ(payload, ToStdVector(data.data)); + std::unique_lock lock{mx}; + numReceivedProxied++; + }); + + directParticipant->Participant()->CreateDataSubscriber("test", pubSubSpec, [&](auto&&, auto&& data) { + EXPECT_EQ(payload, ToStdVector(data.data)); + std::unique_lock lock{mx}; + numReceivedDirect++; + }); + + auto* publisher = publisherParticipant->Participant()->CreateDataPublisher("test", pubSubSpec); + publisherParticipant->GetOrCreateTimeSyncService()->SetSimulationStepHandler([&](auto&&, auto&&) { + std::unique_lock lock{mx}; + if (numReceivedProxied >= 1 && numReceivedDirect >= 1) + { + publisherParticipant->Stop(); + } + else + { + publisher->Publish(payload); + } + }, 1ms); + + ASSERT_TRUE(testSetup.Run(4s)); + EXPECT_GE(numReceivedProxied, 1u); + EXPECT_GE(numReceivedDirect, 1u); +} + } // namespace \ No newline at end of file diff --git a/SilKit/IntegrationTests/ITest_ThreeEthController.cpp b/SilKit/IntegrationTests/ITest_ThreeEthController.cpp index c52012c04..ba9b26634 100644 --- a/SilKit/IntegrationTests/ITest_ThreeEthController.cpp +++ b/SilKit/IntegrationTests/ITest_ThreeEthController.cpp @@ -53,21 +53,31 @@ class ITest_ThreeEthController : public testing::Test protected: ITest_ThreeEthController() { + // pad the message such that the actual frame has exactly the minimum frame length + constexpr size_t PAYLOAD_LENGTH = MINIMUM_ETHERNET_FRAME_LENGTH - 2 * sizeof(EthernetMac) + - sizeof(EthernetEtherType) - sizeof(EthernetVlanTagControlIdentifier) + - sizeof(EthernetEtherType); + MakeTestMessages(PAYLOAD_LENGTH); + + syncParticipantNames = {"EthWriter", "EthReader1", "EthReader2"}; + } + + void MakeTestMessages(size_t payloadLength) + { + testMessages.clear(); testMessages.resize(5); for (auto index = 0u; index < testMessages.size(); index++) { std::stringstream messageBuilder; messageBuilder << "Test Message " << index; std::string messageString = messageBuilder.str(); - // pad the message such that the actual frame has exactly the minimum frame length - constexpr size_t PAYLOAD_LENGTH = MINIMUM_ETHERNET_FRAME_LENGTH - 2 * sizeof(EthernetMac) - - sizeof(EthernetEtherType) - sizeof(EthernetVlanTagControlIdentifier) - - sizeof(EthernetEtherType); - messageString.resize(std::max(messageString.size(), PAYLOAD_LENGTH), ' '); + // NB: pad with a varying pattern, so that shifted or truncated frames are detected + for (size_t i = messageString.size(); i < payloadLength; ++i) + { + messageString.push_back(static_cast('a' + (i + index) % 26)); + } testMessages[index].expectedData = std::move(messageString); } - - syncParticipantNames = {"EthWriter", "EthReader1", "EthReader2"}; } void SetupSender(SilKit::Tests::SimParticipant* participant) @@ -194,4 +204,11 @@ TEST_F(ITest_ThreeEthController, test_eth_ack_callbacks) ExecuteTest(); } +// Frames large enough that the sender serializes each once and shares the body between both readers +TEST_F(ITest_ThreeEthController, test_eth_ack_callbacks_large_frames) +{ + MakeTestMessages(1400); + ExecuteTest(); +} + } // anonymous namespace diff --git a/SilKit/source/config/ParticipantConfiguration.cpp b/SilKit/source/config/ParticipantConfiguration.cpp index 8dd7f5149..d3e19e958 100755 --- a/SilKit/source/config/ParticipantConfiguration.cpp +++ b/SilKit/source/config/ParticipantConfiguration.cpp @@ -176,7 +176,8 @@ bool operator==(const TimeSynchronization& lhs, const TimeSynchronization& rhs) bool operator==(const Experimental& lhs, const Experimental& rhs) { - return lhs.timeSynchronization == rhs.timeSynchronization && lhs.metrics == rhs.metrics; + return lhs.timeSynchronization == rhs.timeSynchronization && lhs.metrics == rhs.metrics + && lhs.useReceiveBufferPool == rhs.useReceiveBufferPool; } bool operator==(const Label& lhs, const Label& rhs) diff --git a/SilKit/source/config/ParticipantConfiguration.hpp b/SilKit/source/config/ParticipantConfiguration.hpp index 195a1f270..8fcd3e1c8 100644 --- a/SilKit/source/config/ParticipantConfiguration.hpp +++ b/SilKit/source/config/ParticipantConfiguration.hpp @@ -326,6 +326,10 @@ struct Experimental { TimeSynchronization timeSynchronization; Metrics metrics; + //! Reuse the buffers that received messages are read into. Disabling this allocates a fresh + //! buffer per received message, so a payload used after its handler returned is a detectable + //! use-after-free instead of silently reading the bytes of a later message. + bool useReceiveBufferPool{true}; }; // ================================================================================ diff --git a/SilKit/source/config/ParticipantConfiguration.schema.json b/SilKit/source/config/ParticipantConfiguration.schema.json index 6094463de..ecd83644b 100644 --- a/SilKit/source/config/ParticipantConfiguration.schema.json +++ b/SilKit/source/config/ParticipantConfiguration.schema.json @@ -1133,6 +1133,12 @@ }, "Metrics": { "$ref": "#/definitions/Metrics" + }, + "UseReceiveBufferPool": { + "type": "boolean", + "description": "Reuse the buffers that received messages are read into. Set to false to allocate a fresh buffer per received message, which turns a payload used after its handler returned into a detectable use-after-free. Intended for diagnosing such bugs.", + "default": true, + "examples": [false] } }, "additionalProperties": false diff --git a/SilKit/source/config/ParticipantConfigurationFromXImpl.cpp b/SilKit/source/config/ParticipantConfigurationFromXImpl.cpp index e26882f73..a16a26c73 100644 --- a/SilKit/source/config/ParticipantConfigurationFromXImpl.cpp +++ b/SilKit/source/config/ParticipantConfigurationFromXImpl.cpp @@ -77,6 +77,7 @@ struct ExperimentalCache { TimeSynchronizationCache timeSynchronizationCache; MetricsCache metricsCache; + std::optional useReceiveBufferPool; }; struct ConfigIncludeData @@ -408,8 +409,11 @@ void Cache(const Metrics& root, MetricsCache& cache) void Cache(const Experimental& root, ExperimentalCache& cache) { + static const Experimental defaultObject; Cache(root.timeSynchronization, cache.timeSynchronizationCache); Cache(root.metrics, cache.metricsCache); + CacheNonDefault(defaultObject.useReceiveBufferPool, root.useReceiveBufferPool, "Experimental.UseReceiveBufferPool", + cache.useReceiveBufferPool); } @@ -557,6 +561,7 @@ void MergeExperimentalCache(const ExperimentalCache& cache, Experimental& experi { MergeTimeSynchronizationCache(cache.timeSynchronizationCache, experimental.timeSynchronization); MergeMetricsCache(cache.metricsCache, experimental.metrics); + MergeCacheField(cache.useReceiveBufferPool, experimental.useReceiveBufferPool); } diff --git a/SilKit/source/config/ParticipantConfiguration_Full.json b/SilKit/source/config/ParticipantConfiguration_Full.json index 8113ce95c..70f14818d 100644 --- a/SilKit/source/config/ParticipantConfiguration_Full.json +++ b/SilKit/source/config/ParticipantConfiguration_Full.json @@ -288,6 +288,7 @@ "Name": "MyRemoteMetricsSink" } ] - } + }, + "UseReceiveBufferPool": false } } \ No newline at end of file diff --git a/SilKit/source/config/ParticipantConfiguration_Full.yaml b/SilKit/source/config/ParticipantConfiguration_Full.yaml index e50498dce..d5696df33 100644 --- a/SilKit/source/config/ParticipantConfiguration_Full.yaml +++ b/SilKit/source/config/ParticipantConfiguration_Full.yaml @@ -205,4 +205,5 @@ Experimental: - Type: JsonFile Name: MyJsonMetrics - Type: Remote - Name: MyRemoteMetricsSink \ No newline at end of file + Name: MyRemoteMetricsSink + UseReceiveBufferPool: false \ No newline at end of file diff --git a/SilKit/source/config/Test_YamlParser.cpp b/SilKit/source/config/Test_YamlParser.cpp index ab5ac1aac..5c87abf29 100644 --- a/SilKit/source/config/Test_YamlParser.cpp +++ b/SilKit/source/config/Test_YamlParser.cpp @@ -414,6 +414,31 @@ TEST_F(Test_YamlParser, yaml_metrics_update_interval) EXPECT_EQ(configDefault.experimental.metrics.updateInterval, 1s); } +TEST_F(Test_YamlParser, yaml_use_receive_buffer_pool) +{ + auto configDefault = Deserialize(R"( +Experimental: + Metrics: + CollectFromRemote: true +)"); + EXPECT_TRUE(configDefault.experimental.useReceiveBufferPool); + + auto config = Deserialize(R"( +Experimental: + UseReceiveBufferPool: false +)"); + EXPECT_FALSE(config.experimental.useReceiveBufferPool); + + auto config2 = Deserialize(Serialize(config)); + EXPECT_FALSE(config2.experimental.useReceiveBufferPool); + + auto configTrue = Deserialize(R"( +Experimental: + UseReceiveBufferPool: true +)"); + EXPECT_TRUE(configTrue.experimental.useReceiveBufferPool); +} + TEST_F(Test_YamlParser, middleware_convert) { auto config = Deserialize(R"( diff --git a/SilKit/source/config/YamlReader.cpp b/SilKit/source/config/YamlReader.cpp index 9d34ace09..bc3b4395a 100644 --- a/SilKit/source/config/YamlReader.cpp +++ b/SilKit/source/config/YamlReader.cpp @@ -501,6 +501,7 @@ void YamlReader::Read(SilKit::Config::Experimental& obj) { OptionalRead(obj.timeSynchronization, "TimeSynchronization"); OptionalRead(obj.metrics, "Metrics"); + OptionalRead(obj.useReceiveBufferPool, "UseReceiveBufferPool"); } void YamlReader::Read(SilKit::Config::ParticipantConfiguration& obj) diff --git a/SilKit/source/config/YamlValidator.cpp b/SilKit/source/config/YamlValidator.cpp index 65dab13d2..554199f86 100644 --- a/SilKit/source/config/YamlValidator.cpp +++ b/SilKit/source/config/YamlValidator.cpp @@ -148,6 +148,7 @@ const std::set schemaPaths_v1 = { "/Experimental/TimeSynchronization/AnimationFactor", "/Experimental/TimeSynchronization/DynamicSimulationStep", "/Experimental/TimeSynchronization/EnableMessageAggregation", + "/Experimental/UseReceiveBufferPool", "/Extensions", "/Extensions/SearchPathHints", "/FlexrayControllers", diff --git a/SilKit/source/config/YamlWriter.cpp b/SilKit/source/config/YamlWriter.cpp index 65cd9de26..0dd96437a 100644 --- a/SilKit/source/config/YamlWriter.cpp +++ b/SilKit/source/config/YamlWriter.cpp @@ -584,6 +584,7 @@ void YamlWriter::Write(const SilKit::Config::Experimental& obj) MakeMap(); NonDefaultWrite(obj.timeSynchronization, "TimeSynchronization", defaultObj.timeSynchronization); NonDefaultWrite(obj.metrics, "Metrics", defaultObj.metrics); + NonDefaultWrite(obj.useReceiveBufferPool, "UseReceiveBufferPool", defaultObj.useReceiveBufferPool); } diff --git a/SilKit/source/core/internal/MessageBuffer.hpp b/SilKit/source/core/internal/MessageBuffer.hpp index 17d0d72ff..96c154806 100644 --- a/SilKit/source/core/internal/MessageBuffer.hpp +++ b/SilKit/source/core/internal/MessageBuffer.hpp @@ -47,6 +47,51 @@ class MessageBufferPeeker }; +//! Maximum number of serialization buffers held per thread, and the largest one worth keeping. +constexpr size_t SerializationBufferPoolSize{32}; +constexpr size_t SerializationBufferPoolMaxCapacity{64 * 1024}; + +namespace Detail { +//! Thread local free list of serialization buffers. See AcquireSerializationBuffer. +inline auto SerializationBufferPool() -> std::vector>& +{ + thread_local std::vector> pool; + return pool; +} +} // namespace Detail + +/*! \brief Take a serialization buffer from the thread local free list, or a fresh one. + * + * A buffer acquired on one thread and recycled on another migrates between lists, which is harmless. + */ +inline auto AcquireSerializationBuffer() -> std::vector +{ + auto& pool = Detail::SerializationBufferPool(); + if (pool.empty()) + { + return {}; + } + + auto buffer = std::move(pool.back()); + pool.pop_back(); + return buffer; +} + +//! Hand a serialization buffer back once its contents are no longer needed. +inline void RecycleSerializationBuffer(std::vector&& buffer) +{ + auto& pool = Detail::SerializationBufferPool(); + if (buffer.capacity() == 0 || buffer.capacity() > SerializationBufferPoolMaxCapacity + || pool.size() >= SerializationBufferPoolSize) + { + return; + } + + buffer.clear(); + pool.push_back(std::move(buffer)); +} + + class MessageBuffer { public: @@ -56,7 +101,11 @@ class MessageBuffer public: // ---------------------------------------- // Constructors and Destructor - inline MessageBuffer() = default; + // NB: the writing constructor takes its storage from the thread local serialization pool. + inline MessageBuffer() + : _storage{AcquireSerializationBuffer()} + { + } inline MessageBuffer(std::vector data); //! \brief Read from a shared blob without copying it. The buffer is read-only in this state. inline explicit MessageBuffer(Util::SharedSpan blob); diff --git a/SilKit/source/core/vasio/CMakeLists.txt b/SilKit/source/core/vasio/CMakeLists.txt index 7384a1b29..89f757f3d 100644 --- a/SilKit/source/core/vasio/CMakeLists.txt +++ b/SilKit/source/core/vasio/CMakeLists.txt @@ -59,6 +59,9 @@ add_library(O_SilKit_Core_VAsio OBJECT RingBuffer.hpp RingBuffer.cpp + ReceiveBlobPool.hpp + ReceiveBlobPool.cpp + IPeerMetrics.hpp PeerMetrics.hpp PeerMetrics.cpp @@ -91,6 +94,9 @@ add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_TransformAcceptorUris add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_VAsioCapabilities.cpp LIBS S_SilKitImpl) add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_RingBuffer.cpp LIBS S_SilKitImpl) +add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_ReceiveBlobPool.cpp LIBS S_SilKitImpl) +add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_VAsioPeer.cpp LIBS S_SilKitImpl I_SilKit) +add_silkit_test_to_executable(SilKitUnitTests SOURCES Test_VAsioProxyPeer.cpp LIBS S_SilKitImpl I_SilKit) add_silkit_test_to_executable(SilKitUnitTests SOURCES io/Test_IoContext.cpp LIBS S_SilKitImpl) add_silkit_test_to_executable(SilKitUnitTests SOURCES io/Test_AsioIoContext.cpp LIBS S_SilKitImpl) diff --git a/SilKit/source/core/vasio/ReceiveBlobPool.cpp b/SilKit/source/core/vasio/ReceiveBlobPool.cpp new file mode 100644 index 000000000..72335452c --- /dev/null +++ b/SilKit/source/core/vasio/ReceiveBlobPool.cpp @@ -0,0 +1,91 @@ +// SPDX-FileCopyrightText: 2026 Vector Informatik GmbH +// +// SPDX-License-Identifier: MIT + +#include "core/vasio/ReceiveBlobPool.hpp" + +#include + +namespace SilKit { +namespace Core { + +ReceiveBlobPool::ReceiveBlobPool(bool enabled) + : _enabled{enabled} +{ +} + +auto ReceiveBlobPool::IsEnabled() const -> bool +{ + return _enabled; +} + +auto ReceiveBlobPool::Acquire(size_t size) -> std::shared_ptr> +{ + if (!_enabled || size > MaxBlobSize) + { + return std::make_shared>(size); + } + + // NB: prefer the smallest free blob that holds the message within the slack, so that a mix of + // message sizes settles into blobs of matching sizes instead of reallocating back and forth. + std::shared_ptr>* fitting{nullptr}; + std::shared_ptr>* tooSmall{nullptr}; + std::shared_ptr>* tooLarge{nullptr}; + + for (auto& entry : _entries) + { + if (entry.use_count() != 1) + { + continue; + } + + const auto entrySize = entry->size(); + if (entrySize < size) + { + if (tooSmall == nullptr || entrySize > (*tooSmall)->size()) + { + tooSmall = &entry; + } + } + else if (entrySize - size > MaxSlack) + { + tooLarge = &entry; + } + else if (fitting == nullptr || entrySize < (*fitting)->size()) + { + fitting = &entry; + } + } + + auto* reused = (fitting != nullptr) ? fitting : tooSmall; + if (reused != nullptr) + { + // NB: use_count() is a relaxed load. The fence orders the writes into the reused blob after + // the reads of whichever thread dropped the second-to-last reference to it. + std::atomic_thread_fence(std::memory_order_acquire); + + // NB: only grow. Shrinking would value initialize bytes that the caller overwrites anyway, + // so the caller views only the part it fills. + if ((*reused)->size() < size) + { + (*reused)->resize(size); + } + return *reused; + } + + auto blob = std::make_shared>(size); + if (_entries.size() < MaxEntries) + { + _entries.push_back(blob); + } + else if (tooLarge != nullptr) + { + // Drop an oversized blob in favour of one that matches the current traffic. + *tooLarge = blob; + } + + return blob; +} + +} // namespace Core +} // namespace SilKit diff --git a/SilKit/source/core/vasio/ReceiveBlobPool.hpp b/SilKit/source/core/vasio/ReceiveBlobPool.hpp new file mode 100644 index 000000000..8d3ba33c1 --- /dev/null +++ b/SilKit/source/core/vasio/ReceiveBlobPool.hpp @@ -0,0 +1,48 @@ +// SPDX-FileCopyrightText: 2026 Vector Informatik GmbH +// +// SPDX-License-Identifier: MIT + +#pragma once + +#include +#include +#include +#include + +namespace SilKit { +namespace Core { + +/*! \brief The blobs that received messages are linearised into. + * + * Deserialized payloads alias their blob, so a blob is only reused once the pool holds the last + * reference. A disabled pool allocates a fresh blob per message, see + * Config::Experimental::useReceiveBufferPool. + * + * Not thread safe. The owning connection only uses it from its io thread. + */ +class ReceiveBlobPool +{ +public: + //! Maximum number of blobs held. + static constexpr size_t MaxEntries{8}; + //! Blobs for larger messages are not pooled. Large messages are rare; do not hoard memory for them. + static constexpr size_t MaxBlobSize{64 * 1024}; + //! A reused blob exceeds the requested size by at most this many bytes. A payload that is retained + //! past its handler keeps its whole blob alive, so this bounds the memory a small payload can pin. + static constexpr size_t MaxSlack{4 * 1024}; + +public: + explicit ReceiveBlobPool(bool enabled); + + //! \brief A blob of at least size bytes with unspecified contents. + auto Acquire(size_t size) -> std::shared_ptr>; + + auto IsEnabled() const -> bool; + +private: + bool _enabled; + std::vector>> _entries; +}; + +} // namespace Core +} // namespace SilKit diff --git a/SilKit/source/core/vasio/SilKitLink.hpp b/SilKit/source/core/vasio/SilKitLink.hpp index 599fe7d01..5808deb8c 100644 --- a/SilKit/source/core/vasio/SilKitLink.hpp +++ b/SilKit/source/core/vasio/SilKitLink.hpp @@ -72,7 +72,14 @@ class SilKitLink Services::Logging::ILoggerInternal* _logger; Services::Orchestration::ITimeProvider* _timeProvider; - std::vector _localReceivers; + //! The service endpoint is resolved on registration, to avoid a dynamic_cast per message. + struct LocalReceiver + { + ReceiverT* receiver; + const IServiceEndpoint* serviceEndpoint; + }; + + std::vector _localReceivers; VAsioTransmitter _vasioTransmitter; }; @@ -93,9 +100,13 @@ SilKitLink::SilKitLink(std::string name, Services::Logging::ILoggerInterna template void SilKitLink::AddLocalReceiver(ReceiverT* receiver) { - if (std::find(_localReceivers.begin(), _localReceivers.end(), receiver) != _localReceivers.end()) + const auto alreadyRegistered = + std::find_if(_localReceivers.begin(), _localReceivers.end(), + [receiver](const LocalReceiver& entry) { return entry.receiver == receiver; }); + if (alreadyRegistered != _localReceivers.end()) return; - _localReceivers.push_back(receiver); + + _localReceivers.push_back(LocalReceiver{receiver, dynamic_cast(receiver)}); } template @@ -152,7 +163,7 @@ void SilKitLink::DistributeRemoteSilKitMessage(const IServiceEndpoint* fro for (auto&& receiver : _localReceivers) { - DispatchSilKitMessage(receiver, from, msg); + DispatchSilKitMessage(receiver.receiver, from, msg); } } @@ -178,7 +189,7 @@ void SilKitLink::DistributeToSelf(const IServiceEndpoint* from, const MsgT { for (auto&& receiver : _localReceivers) { - auto* receiverId = dynamic_cast(receiver); + const auto* receiverId = receiver.serviceEndpoint; if constexpr (!SilKitMsgTraits::IsSelfDeliveryEnforced()) { if (receiverId->GetServiceDescriptor() == from->GetServiceDescriptor()) @@ -187,7 +198,7 @@ void SilKitLink::DistributeToSelf(const IServiceEndpoint* from, const MsgT // Trace reception of self delivery Services::TraceRx(_logger, receiverId, msg, from->GetServiceDescriptor()); - DispatchSilKitMessage(receiver, from, msg); + DispatchSilKitMessage(receiver.receiver, from, msg); } } } diff --git a/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp new file mode 100644 index 000000000..272be789d --- /dev/null +++ b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp @@ -0,0 +1,165 @@ +// SPDX-FileCopyrightText: 2026 Vector Informatik GmbH +// +// SPDX-License-Identifier: MIT + +#include "core/vasio/ReceiveBlobPool.hpp" + +#include +#include +#include +#include + +#include "gtest/gtest.h" + +namespace { + +using SilKit::Core::ReceiveBlobPool; + +using Blob = std::shared_ptr>; +using WeakBlob = std::weak_ptr>; + +constexpr size_t Small{100}; +constexpr size_t Large{60 * 1024}; + +TEST(Test_ReceiveBlobPool, reuses_an_unreferenced_blob) +{ + ReceiveBlobPool pool{true}; + + const auto* first = pool.Acquire(Small).get(); + const auto second = pool.Acquire(Small); + + EXPECT_EQ(second.get(), first); + EXPECT_GE(second->size(), Small); +} + +TEST(Test_ReceiveBlobPool, skips_a_referenced_blob) +{ + ReceiveBlobPool pool{true}; + + const auto retained = pool.Acquire(Small); + const auto second = pool.Acquire(Small); + + EXPECT_NE(second.get(), retained.get()); +} + +TEST(Test_ReceiveBlobPool, grows_a_blob_that_is_too_small) +{ + ReceiveBlobPool pool{true}; + + const auto* first = pool.Acquire(Small).get(); + const auto second = pool.Acquire(10 * Small); + + EXPECT_EQ(second.get(), first); + EXPECT_GE(second->size(), 10 * Small); +} + +TEST(Test_ReceiveBlobPool, does_not_pool_blobs_above_the_limit) +{ + ReceiveBlobPool pool{true}; + + const WeakBlob large = pool.Acquire(ReceiveBlobPool::MaxBlobSize + 1); + + EXPECT_TRUE(large.expired()); +} + +TEST(Test_ReceiveBlobPool, bounds_the_slack_of_a_reused_blob) +{ + ReceiveBlobPool pool{true}; + + const auto* large = pool.Acquire(Large).get(); + const auto small = pool.Acquire(Small); + + EXPECT_NE(small.get(), large); + EXPECT_LE(small->size(), Small + ReceiveBlobPool::MaxSlack); +} + +TEST(Test_ReceiveBlobPool, mixed_sizes_settle_into_matching_blobs) +{ + ReceiveBlobPool pool{true}; + + // The first large message grows the only blob, so the pool settles in the second round. + pool.Acquire(Small); + pool.Acquire(Large); + + const auto* small = pool.Acquire(Small).get(); + const auto* large = pool.Acquire(Large).get(); + EXPECT_NE(small, large); + + for (int i = 0; i < 10; ++i) + { + EXPECT_EQ(pool.Acquire(Small).get(), small); + EXPECT_EQ(pool.Acquire(Large).get(), large); + } +} + +TEST(Test_ReceiveBlobPool, full_pool_hands_out_unpooled_blobs) +{ + ReceiveBlobPool pool{true}; + + std::vector retained; + for (size_t i = 0; i < ReceiveBlobPool::MaxEntries; ++i) + { + retained.push_back(pool.Acquire(Small)); + } + + const WeakBlob extra = pool.Acquire(Small); + + EXPECT_TRUE(extra.expired()); +} + +TEST(Test_ReceiveBlobPool, full_pool_replaces_an_oversized_blob) +{ + ReceiveBlobPool pool{true}; + + std::vector large; + { + std::vector retained; + for (size_t i = 0; i < ReceiveBlobPool::MaxEntries; ++i) + { + retained.push_back(pool.Acquire(Large)); + large.push_back(retained.back()); + } + } + + const WeakBlob small = pool.Acquire(Small); + + EXPECT_FALSE(small.expired()); + EXPECT_EQ(std::count_if(large.begin(), large.end(), [](const WeakBlob& blob) { return blob.expired(); }), 1); +} + +TEST(Test_ReceiveBlobPool, disabled_pool_keeps_no_blobs) +{ + ReceiveBlobPool pool{false}; + + for (int i = 0; i < 3; ++i) + { + const WeakBlob blob = pool.Acquire(Small); + EXPECT_TRUE(blob.expired()); + } +} + +// NB: TSan does not model std::atomic_thread_fence, so under TSan this test needs a suppression for +// ReceiveBlobPool::Acquire. +TEST(Test_ReceiveBlobPool, reuses_a_blob_released_on_another_thread) +{ + ReceiveBlobPool pool{true}; + + auto blob = pool.Acquire(Small); + std::fill(blob->begin(), blob->end(), uint8_t{0x42}); + const auto* first = blob.get(); + + uint8_t observed{0}; + std::thread reader{[alias = std::move(blob), &observed]() mutable { + observed = alias->front(); + alias.reset(); + }}; + reader.join(); + + auto reused = pool.Acquire(Small); + std::fill(reused->begin(), reused->end(), uint8_t{0x17}); + + EXPECT_EQ(observed, 0x42); + EXPECT_EQ(reused.get(), first); +} + +} // namespace diff --git a/SilKit/source/core/vasio/Test_SerializedMessage.cpp b/SilKit/source/core/vasio/Test_SerializedMessage.cpp index d9efd9f20..27cdaa4d2 100644 --- a/SilKit/source/core/vasio/Test_SerializedMessage.cpp +++ b/SilKit/source/core/vasio/Test_SerializedMessage.cpp @@ -3,10 +3,15 @@ // SPDX-License-Identifier: MIT #include "core/vasio/SerializedMessage.hpp" +#include "wire/can/WireCanMessages.hpp" +#include "wire/ethernet/WireEthernetMessages.hpp" +#include "wire/flexray/WireFlexrayMessages.hpp" #include "wire/pubsub/WireDataMessages.hpp" +#include "wire/rpc/WireRpcMessages.hpp" #include #include +#include #include #include #include @@ -218,3 +223,185 @@ TEST(Test_SerializedMessage, a_shared_blob_buffer_rejects_writes) ASSERT_THROW(receiving.ReleaseStorage(), SilKit::SilKitError); } + +namespace { + +auto MakeTestPayload(size_t size) -> std::vector +{ + std::vector payload(size); + for (size_t i = 0; i < size; ++i) + { + payload[i] = static_cast(i * 31 + 7); + } + return payload; +} + +template +void AppendRaw(std::vector& bytes, const T& value) +{ + const auto* first = reinterpret_cast(&value); + bytes.insert(bytes.end(), first, first + sizeof(value)); +} + +void AppendLengthPrefixed(std::vector& bytes, const std::vector& payload) +{ + AppendRaw(bytes, static_cast(payload.size())); + bytes.insert(bytes.end(), payload.begin(), payload.end()); +} + +auto ShareBlob(SerializedMessage sending) -> std::shared_ptr> +{ + return std::make_shared>(sending.ReleaseStorage()); +} + +//! Round-trip msg through a shared blob and check that the payload selected by payloadOf aliases it. +template +void ExpectPayloadAliasesTheBlob(const MsgT& msg, PayloadOf payloadOf) +{ + const auto blob = ShareBlob(SerializedMessage{msg, EndpointAddress{1, 2}, EndpointId{3}}); + + SerializedMessage receiving{SilKit::Util::MakeSharedSpan(blob, 0, blob->size())}; + const auto received = receiving.Deserialize(); + const auto& payload = payloadOf(received); + const auto view = payload.AsSpan(); + + EXPECT_GE(view.data(), blob->data()); + EXPECT_LE(view.data() + view.size(), blob->data() + blob->size()); + EXPECT_TRUE(SilKit::Util::ItemsAreEqual(view, payloadOf(msg).AsSpan())); +} + +} // namespace + +TEST(Test_SerializedMessage, can_frame_payload_aliases_a_shared_blob) +{ + SilKit::Services::Can::WireCanFrameEvent event{}; + event.frame.canId = 0x123; + event.frame.dataField = MakeTestPayload(64); + + ExpectPayloadAliasesTheBlob(event, [](const auto& msg) -> const auto& { return msg.frame.dataField; }); +} + +TEST(Test_SerializedMessage, ethernet_frame_payload_aliases_a_shared_blob) +{ + SilKit::Services::Ethernet::WireEthernetFrameEvent event{}; + event.frame.raw = MakeTestPayload(1500); + + ExpectPayloadAliasesTheBlob(event, [](const auto& msg) -> const auto& { return msg.frame.raw; }); +} + +TEST(Test_SerializedMessage, flexray_frame_payload_aliases_a_shared_blob) +{ + SilKit::Services::Flexray::WireFlexrayFrameEvent event{}; + event.channel = SilKit::Services::Flexray::FlexrayChannel::A; + event.frame.payload = MakeTestPayload(254); + + ExpectPayloadAliasesTheBlob(event, [](const auto& msg) -> const auto& { return msg.frame.payload; }); +} + +TEST(Test_SerializedMessage, flexray_tx_buffer_update_payload_aliases_a_shared_blob) +{ + SilKit::Services::Flexray::WireFlexrayTxBufferUpdate update{}; + update.txBufferIndex = 3; + update.payloadDataValid = true; + update.payload = MakeTestPayload(254); + + ExpectPayloadAliasesTheBlob(update, [](const auto& msg) -> const auto& { return msg.payload; }); +} + +TEST(Test_SerializedMessage, rpc_payloads_alias_a_shared_blob) +{ + SilKit::Services::Rpc::FunctionCall call{std::chrono::nanoseconds{1}, SilKit::Util::Uuid{1, 2}, + MakeTestPayload(300)}; + ExpectPayloadAliasesTheBlob(call, [](const auto& msg) -> const auto& { return msg.data; }); + + SilKit::Services::Rpc::FunctionCallResponse response{ + std::chrono::nanoseconds{2}, SilKit::Util::Uuid{3, 4}, MakeTestPayload(300), + SilKit::Services::Rpc::FunctionCallResponse::Status::Success}; + ExpectPayloadAliasesTheBlob(response, [](const auto& msg) -> const auto& { return msg.data; }); +} + +TEST(Test_SerializedMessage, rpc_wire_format_is_unchanged) +{ + // The payload used to be a std::vector, which was written as a uint32 length prefix + // followed by the bytes. The SharedSpan payload must produce the very same bytes. + using SilKit::Services::Rpc::FunctionCall; + using SilKit::Services::Rpc::FunctionCallResponse; + + const auto payload = MakeTestPayload(17); + const FunctionCall call{std::chrono::nanoseconds{0x0102030405060708}, SilKit::Util::Uuid{0x1111, 0x2222}, + payload}; + const FunctionCallResponse response{std::chrono::nanoseconds{-5}, SilKit::Util::Uuid{0x3333, 0x4444}, payload, + FunctionCallResponse::Status::InternalError}; + + std::vector expectedCall; + AppendRaw(expectedCall, int64_t{0x0102030405060708}); + AppendRaw(expectedCall, uint64_t{0x1111}); + AppendRaw(expectedCall, uint64_t{0x2222}); + AppendLengthPrefixed(expectedCall, payload); + + std::vector expectedResponse; + AppendRaw(expectedResponse, int64_t{-5}); + AppendRaw(expectedResponse, uint64_t{0x3333}); + AppendRaw(expectedResponse, uint64_t{0x4444}); + AppendLengthPrefixed(expectedResponse, payload); + AppendRaw(expectedResponse, static_cast(FunctionCallResponse::Status::InternalError)); + + MessageBuffer callBuffer; + Serialize(callBuffer, call); + EXPECT_EQ(callBuffer.ReleaseStorage(), expectedCall); + + MessageBuffer responseBuffer; + Serialize(responseBuffer, response); + EXPECT_EQ(responseBuffer.ReleaseStorage(), expectedResponse); + + // and bytes written by an older version are read back + MessageBuffer callReader{expectedCall}; + FunctionCall readCall{}; + Deserialize(callReader, readCall); + EXPECT_EQ(readCall, call); + EXPECT_EQ(readCall.timestamp, call.timestamp); + + MessageBuffer responseReader{expectedResponse}; + FunctionCallResponse readResponse{}; + Deserialize(responseReader, readResponse); + EXPECT_EQ(readResponse, response); + EXPECT_EQ(readResponse.timestamp, response.timestamp); +} + +TEST(Test_SerializedMessage, empty_payload_from_a_shared_blob) +{ + SilKit::Services::PubSub::WireDataMessageEvent event{std::chrono::nanoseconds{1}, std::vector{}}; + const auto blob = ShareBlob(SerializedMessage{event, EndpointAddress{1, 2}, EndpointId{3}}); + + SerializedMessage receiving{SilKit::Util::MakeSharedSpan(blob, 0, blob->size())}; + SilKit::Services::PubSub::WireDataMessageEvent received{}; + ASSERT_NO_THROW(received = receiving.Deserialize()); + + EXPECT_TRUE(received.data.empty()); + EXPECT_EQ(received.timestamp, event.timestamp); +} + +TEST(Test_SerializedMessage, payload_ending_at_the_blob_end) +{ + // The payload is the last field of a FunctionCall, so its view ends exactly where the blob ends. + const SilKit::Services::Rpc::FunctionCall call{std::chrono::nanoseconds{1}, SilKit::Util::Uuid{1, 2}, + MakeTestPayload(40)}; + const auto blob = ShareBlob(SerializedMessage{call, EndpointAddress{1, 2}, EndpointId{3}}); + + SerializedMessage receiving{SilKit::Util::MakeSharedSpan(blob, 0, blob->size())}; + const auto received = receiving.Deserialize(); + + EXPECT_EQ(received.data.AsSpan().data() + received.data.size(), blob->data() + blob->size()); + EXPECT_TRUE(SilKit::Util::ItemsAreEqual(received.data.AsSpan(), call.data.AsSpan())); +} + +TEST(Test_SerializedMessage, truncated_payload_throws_end_of_buffer) +{ + const SilKit::Services::Rpc::FunctionCall call{std::chrono::nanoseconds{1}, SilKit::Util::Uuid{1, 2}, + MakeTestPayload(40)}; + const auto blob = ShareBlob(SerializedMessage{call, EndpointAddress{1, 2}, EndpointId{3}}); + + SerializedMessage receiving{SilKit::Util::MakeSharedSpan(blob, 0, blob->size() - 1)}; + + EXPECT_THROW(receiving.Deserialize(), end_of_buffer); +} diff --git a/SilKit/source/core/vasio/Test_VAsioConnection.cpp b/SilKit/source/core/vasio/Test_VAsioConnection.cpp index 7819dcfde..7a5d91be1 100644 --- a/SilKit/source/core/vasio/Test_VAsioConnection.cpp +++ b/SilKit/source/core/vasio/Test_VAsioConnection.cpp @@ -161,6 +161,11 @@ class Test_VAsioConnection : public testing::Test { _connection.RegisterSilKitMsgReceiver(receiver); } + + static auto ReceiveBlobPoolOf(const VAsioConnection& connection) -> const ReceiveBlobPool& + { + return connection._receiveBlobPool; + } }; } // namespace Core @@ -262,3 +267,22 @@ TEST_F(Test_VAsioConnection, DISABLED_versioned_send_testmessage) _connection.OnSocketData(&_from, std::move(buffer)); } + +////////////////////////////////////////////////////////////////////// +// Receive buffer pool configuration +////////////////////////////////////////////////////////////////////// + +TEST_F(Test_VAsioConnection, receive_buffer_pool_is_enabled_by_default) +{ + EXPECT_TRUE(ReceiveBlobPoolOf(_connection).IsEnabled()); +} + +TEST_F(Test_VAsioConnection, receive_buffer_pool_follows_the_configuration) +{ + SilKit::Config::ParticipantConfiguration config; + config.experimental.useReceiveBufferPool = false; + + VAsioConnection connection{nullptr, &_dummyMetricsManager, config, "Test_VAsioConnection", 1, &_timeProvider}; + + EXPECT_FALSE(ReceiveBlobPoolOf(connection).IsEnabled()); +} diff --git a/SilKit/source/core/vasio/Test_VAsioPeer.cpp b/SilKit/source/core/vasio/Test_VAsioPeer.cpp new file mode 100644 index 000000000..89aaf5c51 --- /dev/null +++ b/SilKit/source/core/vasio/Test_VAsioPeer.cpp @@ -0,0 +1,632 @@ +// SPDX-FileCopyrightText: 2026 Vector Informatik GmbH +// +// SPDX-License-Identifier: MIT + +#include "core/vasio/VAsioPeer.hpp" + +#include +#include +#include +#include +#include +#include + +#include "core/internal/OrchestrationDatatypes.hpp" +#include "core/vasio/PeerMetrics.hpp" +#include "core/vasio/ReceiveBlobPool.hpp" +#include "core/vasio/SharedSerializedMessage.hpp" +#include "core/vasio/io/mock/MockIoContext.hpp" +#include "core/vasio/io/mock/MockRawByteStream.hpp" +#include "core/vasio/io/mock/MockTimer.hpp" +#include "services/logging/MockLogger.hpp" +#include "wire/pubsub/WireDataMessages.hpp" + +#include "gtest/gtest.h" +#include "gmock/gmock.h" + +namespace { + +using namespace SilKit::Core; +using SilKit::Services::PubSub::WireDataMessageEvent; +using SilKit::Util::ToStdVector; +using testing::_; +using testing::NiceMock; + +const EndpointAddress From{7, 9}; + +//! Non-uniform, so that shifted, duplicated or skipped bytes are detected. +auto MakePayload(size_t size, uint8_t seed) -> std::vector +{ + std::vector payload(size); + for (size_t i = 0; i < size; ++i) + { + payload[i] = static_cast(i * 31 + seed); + } + return payload; +} + +auto MakeEvent(size_t payloadSize, uint8_t seed) -> WireDataMessageEvent +{ + return WireDataMessageEvent{std::chrono::nanoseconds{seed}, MakePayload(payloadSize, seed)}; +} + +template +auto Reference(const MsgT& msg, EndpointId remoteIdx) -> std::vector +{ + return SerializedMessage{msg, From, remoteIdx}.ReleaseStorage(); +} + +auto Concat(std::initializer_list> parts) -> std::vector +{ + std::vector result; + for (const auto& part : parts) + { + result.insert(result.end(), part.begin(), part.end()); + } + return result; +} + +struct TestPeerListener : IVAsioPeerListener +{ + std::function onData; + + void OnSocketData(IVAsioPeer*, SerializedMessage&& message) override + { + if (onData) + { + onData(message); + } + } + + void OnPeerShutdown(IVAsioPeer*) override {} +}; + +//! A VAsioPeer on top of a mocked stream. Writes and reads complete only when the test says so. +class PeerUnderTest +{ +public: + explicit PeerUnderTest(bool useReceiveBlobPool = true) + : _pool{useReceiveBlobPool} + { + ON_CALL(_ioContext, Dispatch(_)).WillByDefault([](std::function function) { function(); }); + ON_CALL(_ioContext, Post(_)).WillByDefault([](std::function function) { function(); }); + ON_CALL(_ioContext, MakeTimer()).WillByDefault([]() -> std::unique_ptr { + return std::make_unique>(); + }); + + auto stream = std::make_unique>(); + _stream = stream.get(); + ON_CALL(*_stream, SetListener(_)).WillByDefault([this](VSilKit::IRawByteStreamListener& listener) { + _streamListener = &listener; + }); + ON_CALL(*_stream, AsyncWriteSome(_)).WillByDefault([this](VSilKit::ConstBufferSequence sequence) { + _writes.emplace_back(sequence.begin(), sequence.end()); + }); + ON_CALL(*_stream, AsyncReadSome(_)).WillByDefault([this](VSilKit::MutableBufferSequence sequence) { + _reads.emplace_back(sequence.begin(), sequence.end()); + }); + + _peer = std::make_unique(&_listener, &_ioContext, std::move(stream), &_logger, + std::make_unique(), &_pool); + } + + auto Peer() -> VAsioPeer& + { + return *_peer; + } + + auto Listener() -> TestPeerListener& + { + return _listener; + } + + auto WriteCount() const -> size_t + { + return _writes.size(); + } + + auto LastWrite() const -> const std::vector& + { + return _writes.back(); + } + + //! Complete every pending write, the first ones with the given byte counts and the rest in full. + //! Returns the bytes as they appeared on the wire. + auto DrainWrites(std::deque transferred = {}) -> std::vector + { + std::vector wire; + while (_completedWrites < _writes.size()) + { + const auto buffers = _writes[_completedWrites++]; + + size_t total{0}; + for (const auto& buffer : buffers) + { + total += buffer.GetSize(); + } + + size_t count = total; + if (!transferred.empty()) + { + count = (std::min)(transferred.front(), total); + transferred.pop_front(); + } + + auto remaining = count; + for (const auto& buffer : buffers) + { + const auto n = (std::min)(remaining, buffer.GetSize()); + const auto* data = static_cast(buffer.GetData()); + wire.insert(wire.end(), data, data + n); + remaining -= n; + } + + _streamListener->OnAsyncWriteSomeDone(*_stream, count); + } + return wire; + } + + //! Deliver bytes to the peer, the first reads with the given byte counts and the rest as large + //! as the offered read buffers allow. + void Feed(const std::vector& bytes, std::deque chunks = {}) + { + size_t offset{0}; + while (offset < bytes.size()) + { + ASSERT_FALSE(_reads.empty()) << "the peer did not request a read"; + const auto buffers = _reads.back(); + _reads.clear(); + + size_t capacity{0}; + for (const auto& buffer : buffers) + { + capacity += buffer.GetSize(); + } + + auto count = (std::min)(capacity, bytes.size() - offset); + if (!chunks.empty()) + { + count = (std::min)(count, chunks.front()); + chunks.pop_front(); + } + + auto remaining = count; + for (const auto& buffer : buffers) + { + const auto n = (std::min)(remaining, buffer.GetSize()); + std::memcpy(buffer.GetData(), bytes.data() + offset + (count - remaining), n); + remaining -= n; + } + + offset += count; + _streamListener->OnAsyncReadSomeDone(*_stream, count); + } + } + +private: + ReceiveBlobPool _pool; + NiceMock _ioContext; + NiceMock _logger; + TestPeerListener _listener; + + NiceMock* _stream{nullptr}; + VSilKit::IRawByteStreamListener* _streamListener{nullptr}; + std::vector> _writes; + size_t _completedWrites{0}; + std::vector> _reads; + + std::unique_ptr _peer; +}; + +// ================================================================================ +// Sending +// ================================================================================ + +TEST(Test_VAsioPeer, small_message_is_written_as_one_inline_buffer) +{ + PeerUnderTest p; + const auto event = MakeEvent(3, 1); + + p.Peer().SendSilKitMsg(SerializedMessage{event, From, EndpointId{42}}); + + ASSERT_EQ(p.WriteCount(), 1u); + EXPECT_EQ(p.LastWrite().size(), 1u); + EXPECT_EQ(p.DrainWrites(), Reference(event, 42)); +} + +TEST(Test_VAsioPeer, large_unshared_message_is_written_as_one_owned_buffer) +{ + PeerUnderTest p; + const auto event = MakeEvent(1024, 2); + + p.Peer().SendSilKitMsg(SerializedMessage{event, From, EndpointId{42}}); + + ASSERT_EQ(p.WriteCount(), 1u); + EXPECT_EQ(p.LastWrite().size(), 1u); + EXPECT_EQ(p.DrainWrites(), Reference(event, 42)); +} + +TEST(Test_VAsioPeer, small_shared_message_is_inlined_with_the_peers_remote_index) +{ + PeerUnderTest p; + const auto event = MakeEvent(3, 3); + const SharedSerializedMessage shared{event, From}; + + p.Peer().SendSilKitMsg(shared, EndpointId{77}); + + ASSERT_EQ(p.WriteCount(), 1u); + EXPECT_EQ(p.LastWrite().size(), 1u); + EXPECT_EQ(p.DrainWrites(), Reference(event, 77)); +} + +TEST(Test_VAsioPeer, large_shared_message_is_written_as_header_and_shared_body) +{ + PeerUnderTest p; + const auto event = MakeEvent(1024, 4); + const SharedSerializedMessage shared{event, From}; + + p.Peer().SendSilKitMsg(shared, EndpointId{77}); + + ASSERT_EQ(p.WriteCount(), 1u); + ASSERT_EQ(p.LastWrite().size(), 2u); + EXPECT_EQ(p.LastWrite()[0].GetSize(), shared.HeaderSize()); + // the body is written from the shared bytes, not from a copy + EXPECT_EQ(p.LastWrite()[1].GetData(), shared.Body().AsSpan().data()); + EXPECT_EQ(p.DrainWrites(), Reference(event, 77)); +} + +TEST(Test_VAsioPeer, peers_share_the_body_but_write_their_own_remote_index) +{ + PeerUnderTest a; + PeerUnderTest b; + const auto event = MakeEvent(1024, 5); + const SharedSerializedMessage shared{event, From}; + + a.Peer().SendSilKitMsg(shared, EndpointId{5}); + b.Peer().SendSilKitMsg(shared, EndpointId{9}); + + ASSERT_EQ(a.LastWrite().size(), 2u); + ASSERT_EQ(b.LastWrite().size(), 2u); + EXPECT_EQ(a.LastWrite()[1].GetData(), b.LastWrite()[1].GetData()); + EXPECT_EQ(a.DrainWrites(), Reference(event, 5)); + EXPECT_EQ(b.DrainWrites(), Reference(event, 9)); +} + +TEST(Test_VAsioPeer, partial_writes_resume_where_they_stopped) +{ + const auto event = MakeEvent(1024, 6); + const SharedSerializedMessage shared{event, From}; + const auto expected = Reference(event, 77); + const auto header = shared.HeaderSize(); + const auto total = shared.TotalSize(); + + const std::vector> splits{ + {0}, + {1}, + {header - 1}, + {header}, + {header + 1}, + {total - 1}, + {1, 1, 1, header, 0, 7, 100}, + }; + + for (const auto& split : splits) + { + PeerUnderTest p; + p.Peer().SendSilKitMsg(shared, EndpointId{77}); + EXPECT_EQ(p.DrainWrites(split), expected) << "first transfer: " << split.front(); + } +} + +TEST(Test_VAsioPeer, partial_writes_of_an_owned_buffer_resume_where_they_stopped) +{ + PeerUnderTest p; + const auto event = MakeEvent(1024, 7); + + p.Peer().SendSilKitMsg(SerializedMessage{event, From, EndpointId{42}}); + + EXPECT_EQ(p.DrainWrites({1, 500, 0, 3}), Reference(event, 42)); +} + +TEST(Test_VAsioPeer, queued_items_keep_valid_headers) +{ + PeerUnderTest p; + const auto first = MakeEvent(1024, 8); + const auto second = MakeEvent(3, 9); + const auto third = MakeEvent(2048, 10); + const SharedSerializedMessage sharedFirst{first, From}; + const SharedSerializedMessage sharedSecond{second, From}; + const SharedSerializedMessage sharedThird{third, From}; + + p.Peer().SendSilKitMsg(sharedFirst, EndpointId{1}); + p.Peer().SendSilKitMsg(sharedSecond, EndpointId{2}); + p.Peer().SendSilKitMsg(sharedThird, EndpointId{3}); + + // only the first item is in flight, the others wait in the queue + ASSERT_EQ(p.WriteCount(), 1u); + EXPECT_EQ(p.DrainWrites({10}), Concat({Reference(first, 1), Reference(second, 2), Reference(third, 3)})); +} + +TEST(Test_VAsioPeer, queued_items_are_gathered_into_one_write) +{ + PeerUnderTest p; + const auto inFlight = MakeEvent(3, 16); + const auto small = MakeEvent(3, 17); + const auto owned = MakeEvent(1024, 18); + const auto sharedEvent = MakeEvent(1024, 19); + const SharedSerializedMessage shared{sharedEvent, From}; + + p.Peer().SendSilKitMsg(SerializedMessage{inFlight, From, EndpointId{1}}); + p.Peer().SendSilKitMsg(SerializedMessage{small, From, EndpointId{2}}); + p.Peer().SendSilKitMsg(SerializedMessage{owned, From, EndpointId{3}}); + p.Peer().SendSilKitMsg(shared, EndpointId{4}); + + EXPECT_EQ(p.DrainWrites(), Concat({Reference(inFlight, 1), Reference(small, 2), Reference(owned, 3), + Reference(sharedEvent, 4)})); + EXPECT_EQ(p.WriteCount(), 2u); + EXPECT_EQ(p.LastWrite().size(), 4u); +} + +TEST(Test_VAsioPeer, partial_writes_of_gathered_items_resume_where_they_stopped) +{ + const auto inFlight = MakeEvent(3, 20); + const auto small = MakeEvent(3, 21); + const auto owned = MakeEvent(1024, 22); + const auto sharedEvent = MakeEvent(1024, 23); + const SharedSerializedMessage shared{sharedEvent, From}; + const auto expected = Concat( + {Reference(inFlight, 1), Reference(small, 2), Reference(owned, 3), Reference(sharedEvent, 4)}); + const auto inFlightSize = Reference(inFlight, 1).size(); + const auto smallSize = Reference(small, 2).size(); + const auto ownedSize = Reference(owned, 3).size(); + + const std::vector> splits{ + {smallSize - 1}, + {smallSize}, + {smallSize + 1}, + {smallSize + ownedSize}, + {smallSize + ownedSize + shared.HeaderSize() + 1}, + {1, 1, smallSize, 0, ownedSize}, + }; + + for (const auto& split : splits) + { + PeerUnderTest p; + p.Peer().SendSilKitMsg(SerializedMessage{inFlight, From, EndpointId{1}}); + p.Peer().SendSilKitMsg(SerializedMessage{small, From, EndpointId{2}}); + p.Peer().SendSilKitMsg(SerializedMessage{owned, From, EndpointId{3}}); + p.Peer().SendSilKitMsg(shared, EndpointId{4}); + + auto transferred = split; + transferred.push_front(inFlightSize); + EXPECT_EQ(p.DrainWrites(transferred), expected) << "first transfer: " << split.front(); + } +} + +TEST(Test_VAsioPeer, gathering_is_limited_by_the_number_of_items) +{ + PeerUnderTest p; + std::vector expected; + + for (size_t i = 0; i < 1 + 2 * VAsioPeer::MaxItemsPerWrite; ++i) + { + const auto event = MakeEvent(3, static_cast(i)); + p.Peer().SendSilKitMsg(SerializedMessage{event, From, EndpointId{i}}); + const auto reference = Reference(event, EndpointId{i}); + expected.insert(expected.end(), reference.begin(), reference.end()); + } + + EXPECT_EQ(p.DrainWrites(), expected); + // the first item alone, then two full batches + EXPECT_EQ(p.WriteCount(), 3u); +} + +TEST(Test_VAsioPeer, gathering_stops_before_the_byte_limit_is_exceeded) +{ + PeerUnderTest p; + const auto inFlight = MakeEvent(3, 24); + const auto small = MakeEvent(3, 25); + const auto large = MakeEvent(VAsioPeer::MaxBytesPerWrite, 26); + const auto last = MakeEvent(3, 27); + + p.Peer().SendSilKitMsg(SerializedMessage{inFlight, From, EndpointId{1}}); + p.Peer().SendSilKitMsg(SerializedMessage{small, From, EndpointId{2}}); + p.Peer().SendSilKitMsg(SerializedMessage{large, From, EndpointId{3}}); + p.Peer().SendSilKitMsg(SerializedMessage{last, From, EndpointId{4}}); + + EXPECT_EQ(p.DrainWrites(), + Concat({Reference(inFlight, 1), Reference(small, 2), Reference(large, 3), Reference(last, 4)})); + // the large item exceeds the limit on its own, so it is neither joined nor followed by another + EXPECT_EQ(p.WriteCount(), 4u); +} + +TEST(Test_VAsioPeer, body_outlives_the_shared_message) +{ + PeerUnderTest p; + const auto event = MakeEvent(1024, 11); + + { + const SharedSerializedMessage shared{event, From}; + p.Peer().SendSilKitMsg(shared, EndpointId{5}); + } + + EXPECT_EQ(p.DrainWrites(), Reference(event, 5)); +} + +TEST(Test_VAsioPeer, aggregation_concatenates_inline_owned_and_shared_items) +{ + PeerUnderTest p; + p.Peer().EnableAggregation(); + + const auto small = MakeEvent(3, 12); + const auto owned = MakeEvent(1024, 13); + const auto sharedEvent = MakeEvent(2048, 14); + const SharedSerializedMessage shared{sharedEvent, From}; + const SilKit::Services::Orchestration::NextSimTask nextSimTask{std::chrono::nanoseconds{1}, + std::chrono::nanoseconds{2}}; + + p.Peer().SendSilKitMsg(SerializedMessage{small, From, EndpointId{1}}); + p.Peer().SendSilKitMsg(SerializedMessage{owned, From, EndpointId{2}}); + p.Peer().SendSilKitMsg(shared, EndpointId{3}); + EXPECT_EQ(p.WriteCount(), 0u); + + p.Peer().SendSilKitMsg(SerializedMessage{nextSimTask, From, EndpointId{4}}); + + ASSERT_EQ(p.WriteCount(), 1u); + EXPECT_EQ(p.DrainWrites(), Concat({Reference(small, 1), Reference(owned, 2), Reference(sharedEvent, 3), + Reference(nextSimTask, 4)})); +} + +TEST(Test_VAsioPeer, aggregated_messages_are_written_before_a_later_unaggregated_one) +{ + PeerUnderTest p; + p.Peer().EnableAggregation(); + + const auto small = MakeEvent(3, 28); + const auto owned = MakeEvent(1024, 29); + const SilKit::Services::Orchestration::SystemCommand command{ + SilKit::Services::Orchestration::SystemCommand::Kind::AbortSimulation}; + + p.Peer().SendSilKitMsg(SerializedMessage{small, From, EndpointId{1}}); + p.Peer().SendSilKitMsg(SerializedMessage{owned, From, EndpointId{2}}); + p.Peer().SendSilKitMsg(SerializedMessage{command, From, EndpointId{3}}); + + EXPECT_EQ(p.DrainWrites(), Concat({Reference(small, 1), Reference(owned, 2), Reference(command, 3)})); +} + +TEST(Test_VAsioPeer, shutdown_with_a_write_in_flight) +{ + PeerUnderTest p; + const auto event = MakeEvent(1024, 15); + const SharedSerializedMessage shared{event, From}; + + p.Peer().SendSilKitMsg(shared, EndpointId{1}); + p.Peer().SendSilKitMsg(shared, EndpointId{2}); + p.Peer().Shutdown(); + + // the in-flight write completes, the queued one was dropped + EXPECT_EQ(p.DrainWrites(), Reference(event, 1)); + + p.Peer().SendSilKitMsg(shared, EndpointId{3}); + EXPECT_EQ(p.WriteCount(), 1u); +} + +// ================================================================================ +// Receiving, with and without the receive blob pool +// ================================================================================ + +class Test_VAsioPeerReceive : public testing::TestWithParam +{ +protected: + Test_VAsioPeerReceive() + : p{GetParam()} + { + p.Listener().onData = [this](SerializedMessage& message) { + storageSizes.push_back(message.GetStorageSize()); + auto event = message.Deserialize(); + payloads.push_back(ToStdVector(event.data.AsSpan())); + if (retain) + { + retained.push_back(std::move(event)); + } + }; + p.Peer().StartAsyncRead(); + } + + PeerUnderTest p; + bool retain{false}; + std::vector storageSizes; + std::vector> payloads; + std::vector retained; +}; + +TEST_P(Test_VAsioPeerReceive, payload_is_delivered_intact) +{ + const auto event = MakeEvent(1024, 1); + + p.Feed(Reference(event, 3)); + + ASSERT_EQ(payloads.size(), 1u); + EXPECT_EQ(payloads[0], ToStdVector(event.data.AsSpan())); +} + +TEST_P(Test_VAsioPeerReceive, message_wrapping_the_ring_buffer_is_reassembled) +{ + // 41 bytes of headers, length prefix and timestamp around the payload + const auto first = MakeEvent(4000 - 41, 2); + const auto second = MakeEvent(500 - 41, 3); + const auto firstBytes = Reference(first, 3); + ASSERT_EQ(firstBytes.size(), 4000u); + + // NB: the first read leaves 10 bytes of the second message in the 4096 byte ring, so the + // rest of it is read into the wrapped free region + p.Feed(Concat({firstBytes, Reference(second, 3)}), {4010}); + + ASSERT_EQ(payloads.size(), 2u); + EXPECT_EQ(payloads[0], ToStdVector(first.data.AsSpan())); + EXPECT_EQ(payloads[1], ToStdVector(second.data.AsSpan())); +} + +TEST_P(Test_VAsioPeerReceive, retained_payload_is_not_overwritten) +{ + const auto kept = MakeEvent(200, 4); + + retain = true; + p.Feed(Reference(kept, 3)); + retain = false; + + // more messages than the pool holds, so every blob is up for reuse + for (uint8_t seed = 5; seed < 5 + 2 * ReceiveBlobPool::MaxEntries; ++seed) + { + p.Feed(Reference(MakeEvent(200, seed), 3)); + } + + ASSERT_EQ(retained.size(), 1u); + EXPECT_EQ(ToStdVector(retained[0].data.AsSpan()), ToStdVector(kept.data.AsSpan())); +} + +TEST_P(Test_VAsioPeerReceive, message_after_a_larger_one_is_viewed_exactly) +{ + // NB: small enough that the pooled blob of the first message is reused for the second + const auto large = MakeEvent(2 * 1024, 5); + const auto small = MakeEvent(100, 6); + const auto smallBytes = Reference(small, 3); + + p.Feed(Reference(large, 3)); + p.Feed(smallBytes); + + ASSERT_EQ(payloads.size(), 2u); + EXPECT_EQ(storageSizes[1], smallBytes.size()); + EXPECT_EQ(payloads[1], ToStdVector(small.data.AsSpan())); +} + +TEST_P(Test_VAsioPeerReceive, message_above_the_pool_limit) +{ + const auto large = MakeEvent(ReceiveBlobPool::MaxBlobSize + 6 * 1024, 7); + const auto small = MakeEvent(100, 8); + + p.Feed(Reference(large, 3)); + p.Feed(Reference(small, 3)); + + ASSERT_EQ(payloads.size(), 2u); + EXPECT_EQ(payloads[0], ToStdVector(large.data.AsSpan())); + EXPECT_EQ(payloads[1], ToStdVector(small.data.AsSpan())); +} + +TEST_P(Test_VAsioPeerReceive, byte_by_byte_delivery) +{ + const auto event = MakeEvent(300, 9); + const auto bytes = Reference(event, 3); + + p.Feed(bytes, std::deque(bytes.size(), 1)); + + ASSERT_EQ(payloads.size(), 1u); + EXPECT_EQ(payloads[0], ToStdVector(event.data.AsSpan())); +} + +INSTANTIATE_TEST_SUITE_P(ReceiveBlobPool, Test_VAsioPeerReceive, testing::Values(true, false), + [](const testing::TestParamInfo& info) { + return std::string{info.param ? "Enabled" : "Disabled"}; +}); + +} // namespace diff --git a/SilKit/source/core/vasio/Test_VAsioProxyPeer.cpp b/SilKit/source/core/vasio/Test_VAsioProxyPeer.cpp new file mode 100644 index 000000000..71b8cba78 --- /dev/null +++ b/SilKit/source/core/vasio/Test_VAsioProxyPeer.cpp @@ -0,0 +1,98 @@ +// SPDX-FileCopyrightText: 2026 Vector Informatik GmbH +// +// SPDX-License-Identifier: MIT + +#include "core/vasio/VAsioProxyPeer.hpp" + +#include + +#include "core/vasio/SharedSerializedMessage.hpp" +#include "core/vasio/mock/MockVAsioPeer.hpp" +#include "services/logging/MockLogger.hpp" +#include "wire/pubsub/WireDataMessages.hpp" + +#include "gtest/gtest.h" +#include "gmock/gmock.h" + +namespace { + +using namespace SilKit::Core; +using SilKit::Services::PubSub::WireDataMessageEvent; +using testing::_; +using testing::NiceMock; + +struct NullPeerListener : IVAsioPeerListener +{ + void OnSocketData(IVAsioPeer*, SerializedMessage&&) override {} + void OnPeerShutdown(IVAsioPeer*) override {} +}; + +struct Forwarded +{ + ProxyMessage message; + MessageAggregationKind aggregationKind; +}; + +class Test_VAsioProxyPeer : public testing::Test +{ +protected: + Test_VAsioProxyPeer() + { + registryInfo.participantName = "Registry"; + destinationInfo.participantName = "Destination"; + + ON_CALL(registry, GetInfo()).WillByDefault(testing::ReturnRef(registryInfo)); + ON_CALL(registry, SendSilKitMsg(testing::Matcher(_))) + .WillByDefault([this](SerializedMessage message) { + const auto aggregationKind = message.GetAggregationKind(); + forwarded.push_back(Forwarded{message.Deserialize(), aggregationKind}); + }); + + proxy = std::make_unique(&listener, "Source", destinationInfo, ®istry, &logger); + } + + const EndpointAddress from{7, 9}; + + NiceMock logger; + NullPeerListener listener; + VAsioPeerInfo registryInfo; + VAsioPeerInfo destinationInfo; + NiceMock registry; + std::unique_ptr proxy; + std::vector forwarded; +}; + +TEST_F(Test_VAsioProxyPeer, shared_and_plain_send_forward_the_same_payload) +{ + for (const size_t payloadSize : {size_t{3}, size_t{1024}}) + { + forwarded.clear(); + + std::vector payload(payloadSize); + for (size_t i = 0; i < payloadSize; ++i) + { + payload[i] = static_cast(i * 31 + 1); + } + const WireDataMessageEvent event{std::chrono::nanoseconds{5}, payload}; + const SharedSerializedMessage shared{event, from}; + + proxy->SendSilKitMsg(shared, EndpointId{77}); + proxy->SendSilKitMsg(SerializedMessage{event, from, EndpointId{77}}); + + ASSERT_EQ(forwarded.size(), 2u); + const auto& viaShared = forwarded[0]; + const auto& viaPlain = forwarded[1]; + + EXPECT_EQ(viaShared.message.payload, SerializedMessage(event, from, EndpointId{77}).ReleaseStorage()) + << "payload size " << payloadSize; + EXPECT_EQ(viaShared.message.payload, viaPlain.message.payload); + EXPECT_EQ(viaShared.message.source, "Source"); + EXPECT_EQ(viaShared.message.destination, "Destination"); + EXPECT_EQ(viaShared.message.source, viaPlain.message.source); + EXPECT_EQ(viaShared.message.destination, viaPlain.message.destination); + EXPECT_EQ(viaShared.aggregationKind, MessageAggregationKind::UserDataMessage); + EXPECT_EQ(viaShared.aggregationKind, viaPlain.aggregationKind); + } +} + +} // namespace diff --git a/SilKit/source/core/vasio/VAsioConnection.cpp b/SilKit/source/core/vasio/VAsioConnection.cpp index 63ebef453..1b61bfcd8 100644 --- a/SilKit/source/core/vasio/VAsioConnection.cpp +++ b/SilKit/source/core/vasio/VAsioConnection.cpp @@ -233,6 +233,7 @@ VAsioConnection::VAsioConnection(IParticipantInternal* participant, IMetricsMana ParticipantId participantId, Services::Orchestration::ITimeProvider* timeProvider, ProtocolVersion version) : _config{std::move(config)} + , _receiveBlobPool{_config.experimental.useReceiveBufferPool} , _participantName{std::move(participantName)} , _participantId{participantId} , _timeProvider{timeProvider} @@ -558,6 +559,13 @@ void VAsioConnection::ConnectParticipantToRegistryAndStartIoWorker(const std::st _registry = MakeVAsioPeer(std::move(registryStream)); _registry->SetInfo(registryPeerInfo); + if (!_receiveBlobPool.IsEnabled()) + { + _logger->MakeMessage(Log::Level::Info, TopicOf(*this)) + .SetMessage("Receive buffer pool is disabled (Experimental.UseReceiveBufferPool)") + .Dispatch(); + } + _logger->MakeMessage(Log::Level::Info, TopicOf(*this)) .SetMessage("Connected to registry at '{}' via '{}' ({})", _registry->GetRemoteAddress(), _registry->GetLocalAddress(), @@ -1783,11 +1791,10 @@ void VAsioConnection::ReceiveRawSilKitMessage(IVAsioPeer* from, SerializedMessag auto endpoint = buffer.GetEndpointAddress(); //ExtractEndpointAddress(buffer); - auto* fromService = dynamic_cast(from); - ServiceDescriptor tmpService(fromService->GetServiceDescriptor()); - tmpService.SetServiceId(endpoint.endpoint); + _receiveServiceDescriptor = from->GetServiceDescriptor(); + _receiveServiceDescriptor.SetServiceId(endpoint.endpoint); - _vasioReceivers[receiverIdx]->ReceiveRawMsg(from, tmpService, std::move(buffer)); + _vasioReceivers[receiverIdx]->ReceiveRawMsg(from, _receiveServiceDescriptor, std::move(buffer)); } void VAsioConnection::RegisterMessageReceiver(std::function callback) @@ -2026,12 +2033,12 @@ auto VAsioConnection::MakeVAsioPeer(std::unique_ptr stream) -> s if (_config.experimental.metrics.sinks.empty()) { return std::make_unique(this, _ioContext.get(), std::move(stream), _logger, - std::make_unique()); + std::make_unique(), &_receiveBlobPool); } else { return std::make_unique(this, _ioContext.get(), std::move(stream), _logger, - std::make_unique()); + std::make_unique(), &_receiveBlobPool); } } diff --git a/SilKit/source/core/vasio/VAsioConnection.hpp b/SilKit/source/core/vasio/VAsioConnection.hpp index daaf2a6c4..5906d1d72 100644 --- a/SilKit/source/core/vasio/VAsioConnection.hpp +++ b/SilKit/source/core/vasio/VAsioConnection.hpp @@ -42,6 +42,7 @@ #include "util/Assert.hpp" #include "services/logging/LoggerMessage.hpp" #include "core/vasio/VAsioCapabilities.hpp" +#include "core/vasio/ReceiveBlobPool.hpp" #include "wire/lin/WireLinMessages.hpp" #include "util/Uri.hpp" #include "services/metrics/Metrics.hpp" @@ -494,6 +495,8 @@ class VAsioConnection // ---------------------------------------- // private members SilKit::Config::ParticipantConfiguration _config; + // NB: declared early so that it outlives the peers, which read received messages into it. + ReceiveBlobPool _receiveBlobPool; std::string _participantName; ParticipantId _participantId{0}; Services::Logging::ILoggerInternal* _logger{nullptr}; @@ -509,6 +512,10 @@ class VAsioConnection //! \brief Lookup for links by name. Util::tuple_tools::wrapped_tuple _serviceToLinkMap; + //! Reused for every received message, to avoid reallocating the descriptor's strings and + //! supplemental data map each time. Only touched on the io thread, and dispatch is synchronous. + ServiceDescriptor _receiveServiceDescriptor; + std::vector> _vasioReceivers; std::unordered_set _vasioUniqueReceiverIds; diff --git a/SilKit/source/core/vasio/VAsioPeer.cpp b/SilKit/source/core/vasio/VAsioPeer.cpp index 99083854f..62f4922d4 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -33,12 +33,14 @@ namespace SilKit { namespace Core { VAsioPeer::VAsioPeer(IVAsioPeerListener* listener, IIoContext* ioContext, std::unique_ptr stream, - Services::Logging::ILoggerInternal* logger, std::unique_ptr peerMetrics) + Services::Logging::ILoggerInternal* logger, std::unique_ptr peerMetrics, + ReceiveBlobPool* receiveBlobPool) : _listener{listener} , _ioContext{ioContext} , _socket{std::move(stream)} , _logger{logger} , _msgBuffer{4096} + , _receiveBlobPool{receiveBlobPool} , _peerMetrics{std::move(peerMetrics)} { _socket->SetListener(*this); @@ -152,6 +154,11 @@ void VAsioPeer::DispatchSendItem(SendItem item, MessageAggregationKind aggregati } else { + // NB: must not overtake the aggregated messages sent before it + if (_useAggregation && !_aggregatedMessages.empty()) + { + Flush(); + } EnqueueSendItem(std::move(item)); } } @@ -164,6 +171,7 @@ auto VAsioPeer::MakeSendItem(std::vector blob) -> SendItem { item.inlineSize = blob.size(); std::memcpy(item.inlineData.data(), blob.data(), item.inlineSize); + RecycleSerializationBuffer(std::move(blob)); } else { @@ -191,26 +199,33 @@ void VAsioPeer::EnqueueSendItem(SendItem item) } } +auto VAsioPeer::SendItem::Size() const -> size_t +{ + return inlineSize + ownedBody.size() + sharedBody.size(); +} + void VAsioPeer::BuildCurrentSendingBuffers() { _currentSendingBuffers.clear(); - if (_currentSendItem.inlineSize > 0) + for (const auto& item : _currentSendItems) { - _currentSendingBuffers.emplace_back(_currentSendItem.inlineData.data(), _currentSendItem.inlineSize); - } + if (item.inlineSize > 0) + { + _currentSendingBuffers.emplace_back(item.inlineData.data(), item.inlineSize); + } - if (!_currentSendItem.ownedBody.empty()) - { - _currentSendingBuffers.emplace_back(_currentSendItem.ownedBody.data(), - _currentSendItem.ownedBody.size()); - } - else - { - const auto body = _currentSendItem.sharedBody.AsSpan(); - if (!body.empty()) + if (!item.ownedBody.empty()) + { + _currentSendingBuffers.emplace_back(item.ownedBody.data(), item.ownedBody.size()); + } + else { - _currentSendingBuffers.emplace_back(body.data(), body.size()); + const auto body = item.sharedBody.AsSpan(); + if (!body.empty()) + { + _currentSendingBuffers.emplace_back(body.data(), body.size()); + } } } } @@ -276,8 +291,14 @@ void VAsioPeer::StartAsyncWrite() _sending = true; - _currentSendItem = std::move(_sendingQueue.front()); - _sendingQueue.pop_front(); + size_t size{0}; + do + { + size += _sendingQueue.front().Size(); + _currentSendItems.push_back(std::move(_sendingQueue.front())); + _sendingQueue.pop_front(); + } while (!_sendingQueue.empty() && _currentSendItems.size() < MaxItemsPerWrite + && size + _sendingQueue.front().Size() <= MaxBytesPerWrite); lock.unlock(); BuildCurrentSendingBuffers(); @@ -370,13 +391,15 @@ void VAsioPeer::DispatchBuffer() else { // NB: linearised into a shared blob, so that deserialized payloads can alias it. - auto currentMsg = std::make_shared>(_currentMsgSize); - if (!_msgBuffer.Read(SilKit::Util::ToSpan(*currentMsg))) + const size_t blobSize = _currentMsgSize; + auto currentMsg = _receiveBlobPool->Acquire(blobSize); + + // NB: a pooled blob may be larger than this message, so read and view exactly the + // message's bytes rather than the whole buffer. + if (!_msgBuffer.Read(SilKit::Util::Span{currentMsg->data(), blobSize})) { throw SilKitError("Reading data from ring buffer failed."); } - - const auto blobSize = currentMsg->size(); SerializedMessage message{ SilKit::Util::MakeSharedSpan(std::shared_ptr>{std::move(currentMsg)}, 0, blobSize)}; @@ -436,8 +459,8 @@ void VAsioPeer::OnAsyncWriteSomeDone(IRawByteStream& stream, size_t bytesTransfe return; } - // release the body as soon as it has been written - _currentSendItem = SendItem{}; + // release the bodies as soon as they have been written + _currentSendItems.clear(); _sending = false; StartAsyncWrite(); } diff --git a/SilKit/source/core/vasio/VAsioPeer.hpp b/SilKit/source/core/vasio/VAsioPeer.hpp index e80a3fe4b..07d085cb5 100644 --- a/SilKit/source/core/vasio/VAsioPeer.hpp +++ b/SilKit/source/core/vasio/VAsioPeer.hpp @@ -18,6 +18,7 @@ #include "core/internal/EndpointAddress.hpp" #include "core/internal/MessageBuffer.hpp" #include "core/vasio/RingBuffer.hpp" +#include "core/vasio/ReceiveBlobPool.hpp" #include "core/vasio/VAsioPeerInfo.hpp" #include "core/internal/ProtocolVersion.hpp" @@ -39,6 +40,10 @@ class VAsioPeer // ---------------------------------------- // Public Data Types + //! Limits for gathering queued messages into one write. A single larger message is written alone. + static constexpr size_t MaxItemsPerWrite{16}; + static constexpr size_t MaxBytesPerWrite{64 * 1024}; + public: // ---------------------------------------- // Constructors and Destructor @@ -49,8 +54,10 @@ class VAsioPeer VAsioPeer& operator=(const VAsioPeer& other) = delete; VAsioPeer& operator=(VAsioPeer&& other) = delete; //implicitly deleted because of mutex + //! receiveBlobPool must outlive the peer and is only used on the io thread. VAsioPeer(IVAsioPeerListener* listener, IIoContext* ioContext, std::unique_ptr stream, - Services::Logging::ILoggerInternal* logger, std::unique_ptr metrics); + Services::Logging::ILoggerInternal* logger, std::unique_ptr metrics, + ReceiveBlobPool* receiveBlobPool); ~VAsioPeer() override; @@ -103,6 +110,8 @@ class VAsioPeer std::vector ownedBody; //! Shared with the other peers this message was sent to. SilKit::Util::SharedSpan sharedBody; + + auto Size() const -> size_t; }; void StartAsyncWrite(); @@ -142,13 +151,14 @@ class VAsioPeer std::atomic _currentMsgSize{0u}; RingBuffer _msgBuffer; std::vector _currentReceivingBuffers; + ReceiveBlobPool* _receiveBlobPool{nullptr}; // sending mutable std::mutex _sendingQueueMutex; std::deque _sendingQueue; - // NB: _currentSendingBuffers points into _currentSendItem, including its inline array, so move - // the item into place before building the buffers. - SendItem _currentSendItem; + // NB: _currentSendingBuffers points into _currentSendItems, including their inline arrays, so + // move all items into place before building the buffers. + std::vector _currentSendItems; std::vector _currentSendingBuffers; std::vector _aggregatedMessages; diff --git a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp index 0a1e481f6..2afc2c20c 100644 --- a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp +++ b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp @@ -119,14 +119,23 @@ void AsioGenericRawByteStream::AsyncReadSome(MutableBufferSequence bufferSequenc _reading = true; } - _readBufferSequence.resize(bufferSequence.size()); - std::transform(bufferSequence.begin(), bufferSequence.end(), _readBufferSequence.begin(), - [](const MutableBuffer& buffer) -> asio::mutable_buffer { - return asio::mutable_buffer{buffer.GetData(), buffer.GetSize()}; - }); - - _socket.async_read_some(_readBufferSequence, - [this](const auto& e, auto s) { OnAsioAsyncReadSomeComplete(e, s); }); + // NB: see AsyncWriteSome. Two buffers are only needed when the ring buffer wraps. + if (bufferSequence.size() == 1) + { + _socket.async_read_some(asio::mutable_buffer{bufferSequence[0].GetData(), bufferSequence[0].GetSize()}, + [this](const auto& e, auto s) { OnAsioAsyncReadSomeComplete(e, s); }); + } + else + { + _readBufferSequence.resize(bufferSequence.size()); + std::transform(bufferSequence.begin(), bufferSequence.end(), _readBufferSequence.begin(), + [](const MutableBuffer& buffer) -> asio::mutable_buffer { + return asio::mutable_buffer{buffer.GetData(), buffer.GetSize()}; + }); + + _socket.async_read_some(_readBufferSequence, + [this](const auto& e, auto s) { OnAsioAsyncReadSomeComplete(e, s); }); + } } } @@ -153,14 +162,24 @@ void AsioGenericRawByteStream::AsyncWriteSome(ConstBufferSequence bufferSequence _writing = true; } - _writeBufferSequence.resize(bufferSequence.size()); - std::transform(bufferSequence.begin(), bufferSequence.end(), _writeBufferSequence.begin(), - [](const ConstBuffer& buffer) -> asio::const_buffer { - return asio::const_buffer{buffer.GetData(), buffer.GetSize()}; - }); + // NB: asio allocates to copy a buffer sequence into the operation, but stores a single + // buffer by value. A single buffer is the common case. + if (bufferSequence.size() == 1) + { + _socket.async_write_some(asio::const_buffer{bufferSequence[0].GetData(), bufferSequence[0].GetSize()}, + [this](const auto& e, auto s) { OnAsioAsyncWriteSomeComplete(e, s); }); + } + else + { + _writeBufferSequence.resize(bufferSequence.size()); + std::transform(bufferSequence.begin(), bufferSequence.end(), _writeBufferSequence.begin(), + [](const ConstBuffer& buffer) -> asio::const_buffer { + return asio::const_buffer{buffer.GetData(), buffer.GetSize()}; + }); - _socket.async_write_some(_writeBufferSequence, - [this](const auto& e, auto s) { OnAsioAsyncWriteSomeComplete(e, s); }); + _socket.async_write_some(_writeBufferSequence, + [this](const auto& e, auto s) { OnAsioAsyncWriteSomeComplete(e, s); }); + } } } diff --git a/docs/configuration/experimental-configuration.rst b/docs/configuration/experimental-configuration.rst index 91bf3acd5..ad73b7f1a 100644 --- a/docs/configuration/experimental-configuration.rst +++ b/docs/configuration/experimental-configuration.rst @@ -89,6 +89,34 @@ TimeSynchronization In the case of option *On*, however, it is necessary to verify that the transmission of messages within a time step does not depend on incoming messages from other participants. In this case, the time step will not be terminated and the communication will block. +UseReceiveBufferPool +-------------------- + +.. code-block:: yaml + + Experimental: + UseReceiveBufferPool: false + +.. list-table:: Receive Buffer Pool Configuration + :widths: 15 85 + :header-rows: 1 + + * - Property Name + - Description + + * - UseReceiveBufferPool + - Reuse the buffers that received messages are read into. Defaults to ``true``. + + Payloads passed to message handlers, for example ``DataMessageEvent::data``, are only valid until + the handler returns. With the pool enabled, a payload that is used after its handler returned + silently reads the bytes of a later message. Setting this option to ``false`` allocates a fresh + buffer per received message instead, so such a use becomes a use-after-free, which crashes more + reliably and is reported by memory checkers such as AddressSanitizer. + + .. note:: + This option is intended for diagnosing payload lifetime issues and costs one allocation per + received message. When it is disabled, the participant logs this at ``Info`` level. + Metrics for participants ------------------------ Each participant supports collecting static attributes of a simulation and