core: optimize away dynamic_cast, avoid allocations

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"

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<std::vector<uint8_t>> expectedDataUnordered;
     for (uint8_t d = 0; d < numMsgToPublish; d++)
     {
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
     }
     pubsubs.push_back({"Sub1",
                        {},
@@ -246,8 +286,8 @@ TEST_F(ITest_Internals_DataPubSub, test_2pub_2sub_sync_sametopic)
     std::vector<std::vector<uint8_t>> expectedDataUnordered;
     for (uint8_t d = 0; d < numMsgToPublish; d++)
     {
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
     }
 
     pubsubs.push_back({"Sub1",
@@ -299,9 +339,9 @@ TEST_F(ITest_Internals_DataPubSub, test_3pub_4sub_sync_4topics)
     std::vector<std::vector<uint8_t>> expectedDataUnordered;
     for (uint8_t d = 0; d < numMsgToPublish; d++)
     {
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
     }
     pubsubs.push_back({"Sub1",
                        {},
@@ -698,8 +738,8 @@ TEST_F(ITest_Internals_DataPubSub, test_1_participant_selfdelivery_same_topic)
     std::vector<std::vector<uint8_t>> expectedDataUnordered;
     for (uint8_t d = 0; d < numMsgToPublish; d++)
     {
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
-        expectedDataUnordered.emplace_back(std::vector<uint8_t>(defaultMsgSize, d));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(d)));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, static_cast<uint8_t>(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<uint8_t>(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<uint8_t>(defaultMsgSize, 0));
+        expectedDataUnordered.emplace_back(MakeTestData(defaultMsgSize, 0));
     }
     subscribers.push_back(
         {"Sub1",

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<uint8_t>
+{
+    std::vector<uint8_t> data(size);
+    for (size_t i = 0; i < size; ++i)
+    {
+        data[i] = static_cast<uint8_t>(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<uint8_t>(messageSizeInBytes, static_cast<uint8_t>(publishMsgCounter));
+                auto data = MakeTestData(messageSizeInBytes, static_cast<uint8_t>(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<uint8_t>(messageSizeInBytes, static_cast<uint8_t>(receiveMsgCounter));
+                    auto expectedData = MakeTestData(messageSizeInBytes, static_cast<uint8_t>(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<SilKit::Config::IParticipantConfiguration> {
+            if (participantConfig.empty())
+            {
+                return SilKit::Config::MakeEmptyParticipantConfigurationImpl();
+            }
+            return SilKit::Config::ParticipantConfigurationFromStringImpl(participantConfig);
+        };
+
+        std::vector<PubSubParticipant> 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<PubSubParticipant>& publishers, std::vector<PubSubParticipant>& subscribers)
     {
         _testSystem.SetupRegistryAndSystemMaster("silkit://localhost:0", false, {});

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<std::vector<uint8_t>> expectedReturnDataUnordered;
+    for (uint8_t d = 0; d < defaultNumCalls; d++)
+    {
+        expectedReturnDataUnordered.emplace_back(std::vector<uint8_t>(messageSize, d + rpcFuncIncrement));
+        expectedReturnDataUnordered.emplace_back(std::vector<uint8_t>(messageSize, d + rpcFuncIncrement));
+    }
+
+    std::vector<RpcParticipant> 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)
 {

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<uint8_t> payload(4096);
+    for (size_t i = 0; i < payload.size(); ++i)
+    {
+        payload[i] = static_cast<uint8_t>(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<decltype(mx)> lock{mx};
+        numReceivedProxied++;
+    });
+
+    directParticipant->Participant()->CreateDataSubscriber("test", pubSubSpec, [&](auto&&, auto&& data) {
+        EXPECT_EQ(payload, ToStdVector(data.data));
+        std::unique_lock<decltype(mx)> lock{mx};
+        numReceivedDirect++;
+    });
+
+    auto* publisher = publisherParticipant->Participant()->CreateDataPublisher("test", pubSubSpec);
+    publisherParticipant->GetOrCreateTimeSyncService()->SetSimulationStepHandler([&](auto&&, auto&&) {
+        std::unique_lock<decltype(mx)> 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

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<size_t>(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<char>('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

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)