From d466733972d39e78b1558d2236d7aed58fe6d8c5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Tue, 15 Sep 2026 16:43:33 +0200 Subject: [PATCH 1/6] get rid of dynamic_cast in the hot path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marius Börschig --- SilKit/source/core/vasio/SilKitLink.hpp | 25 +++++++++++++++----- SilKit/source/core/vasio/VAsioConnection.cpp | 5 +++- 2 files changed, 23 insertions(+), 7 deletions(-) diff --git a/SilKit/source/core/vasio/SilKitLink.hpp b/SilKit/source/core/vasio/SilKitLink.hpp index 599fe7d01..d110a47a6 100644 --- a/SilKit/source/core/vasio/SilKitLink.hpp +++ b/SilKit/source/core/vasio/SilKitLink.hpp @@ -72,7 +72,16 @@ class SilKitLink Services::Logging::ILoggerInternal* _logger; Services::Orchestration::ITimeProvider* _timeProvider; - std::vector _localReceivers; + //! A local receiver together with its service endpoint. The endpoint is resolved once when the + //! receiver is registered, because resolving it per message required a dynamic_cast, and these + //! receivers use multiple inheritance so that cast has to walk the inheritance graph. + struct LocalReceiver + { + ReceiverT* receiver; + const IServiceEndpoint* serviceEndpoint; + }; + + std::vector _localReceivers; VAsioTransmitter _vasioTransmitter; }; @@ -93,9 +102,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 +165,7 @@ void SilKitLink::DistributeRemoteSilKitMessage(const IServiceEndpoint* fro for (auto&& receiver : _localReceivers) { - DispatchSilKitMessage(receiver, from, msg); + DispatchSilKitMessage(receiver.receiver, from, msg); } } @@ -178,7 +191,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 +200,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/VAsioConnection.cpp b/SilKit/source/core/vasio/VAsioConnection.cpp index 63ebef453..06b77fc8f 100644 --- a/SilKit/source/core/vasio/VAsioConnection.cpp +++ b/SilKit/source/core/vasio/VAsioConnection.cpp @@ -1783,7 +1783,10 @@ void VAsioConnection::ReceiveRawSilKitMessage(IVAsioPeer* from, SerializedMessag auto endpoint = buffer.GetEndpointAddress(); //ExtractEndpointAddress(buffer); - auto* fromService = dynamic_cast(from); + // NB: IVAsioPeer derives publicly from IServiceEndpoint, so this is a plain upcast. It used + // to be a dynamic_cast, which profiling showed to be a measurable share of the work done + // for every received message. + const IServiceEndpoint* fromService = from; ServiceDescriptor tmpService(fromService->GetServiceDescriptor()); tmpService.SetServiceId(endpoint.endpoint); From 1376c446d7df7c9b130a38a359f04df90a91fef1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Tue, 15 Sep 2026 17:51:06 +0200 Subject: [PATCH 2/6] core: optimize away unnecessary copies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reusable receive descriptor — VAsioConnection now assigns into a member ServiceDescriptor instead of constructing a fresh one per received message. std::map assignment reuses nodes and std::string assignment reuses buffers, so the supplemental-data allocations disappear. Safe because only the io thread reaches it and dispatch is synchronous. Pooled serialization buffer — a thread-local free list in MessageBuffer. The writing constructor takes a buffer from it; MakeSendItem hands it back once a small message has been copied inline. Thread-local means no synchronization, and a buffer acquired on one thread and recycled on another just migrates between lists. Pooled receive blob — AcquireReceiveBlob reuses a blob only when its use_count() == 1, which is precisely the condition that no deserialized payload still aliases it. Self-correcting: if a payload somehow outlives dispatch, that entry is skipped and a fresh blob is allocated. Both pools are capped (32 buffers / 8 blobs, 64 KB each) so large messages don't cause hoarding. Signed-off-by: Marius Börschig --- SilKit/source/core/internal/MessageBuffer.hpp | 57 ++++++++++++++++++- SilKit/source/core/vasio/VAsioConnection.cpp | 12 +++- SilKit/source/core/vasio/VAsioConnection.hpp | 4 ++ SilKit/source/core/vasio/VAsioPeer.cpp | 55 +++++++++++++++++- 4 files changed, 122 insertions(+), 6 deletions(-) diff --git a/SilKit/source/core/internal/MessageBuffer.hpp b/SilKit/source/core/internal/MessageBuffer.hpp index 17d0d72ff..d79666896 100644 --- a/SilKit/source/core/internal/MessageBuffer.hpp +++ b/SilKit/source/core/internal/MessageBuffer.hpp @@ -47,6 +47,55 @@ 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. + * + * Serializing a message otherwise allocates a buffer every time and retires it again almost + * immediately, which heap profiling showed as one allocation per sent message. The list is thread + * local so it needs no synchronization; a buffer acquired on one thread and recycled on another + * simply 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(); + buffer.clear(); // keeps the capacity, which is the point of pooling it + 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 +105,13 @@ class MessageBuffer public: // ---------------------------------------- // Constructors and Destructor - inline MessageBuffer() = default; + //! NB: takes its storage from the thread local pool, so serializing a message need not + //! allocate one. Only this writing constructor does so; the reading constructors below + //! bring their own storage. + 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/VAsioConnection.cpp b/SilKit/source/core/vasio/VAsioConnection.cpp index 06b77fc8f..604790998 100644 --- a/SilKit/source/core/vasio/VAsioConnection.cpp +++ b/SilKit/source/core/vasio/VAsioConnection.cpp @@ -1787,10 +1787,16 @@ void VAsioConnection::ReceiveRawSilKitMessage(IVAsioPeer* from, SerializedMessag // to be a dynamic_cast, which profiling showed to be a measurable share of the work done // for every received message. const IServiceEndpoint* fromService = from; - ServiceDescriptor tmpService(fromService->GetServiceDescriptor()); - tmpService.SetServiceId(endpoint.endpoint); - _vasioReceivers[receiverIdx]->ReceiveRawMsg(from, tmpService, std::move(buffer)); + // NB: assign into a reusable descriptor rather than constructing a fresh one. The copy exists + // only to override the service id, and constructing it allocated the nodes of the + // supplemental data map for every received message. Assignment reuses the existing nodes + // and string buffers. Only the io thread reaches this and the dispatch below is + // synchronous, so one scratch instance suffices. + _receiveServiceDescriptor = fromService->GetServiceDescriptor(); + _receiveServiceDescriptor.SetServiceId(endpoint.endpoint); + + _vasioReceivers[receiverIdx]->ReceiveRawMsg(from, _receiveServiceDescriptor, std::move(buffer)); } void VAsioConnection::RegisterMessageReceiver(std::function callback) diff --git a/SilKit/source/core/vasio/VAsioConnection.hpp b/SilKit/source/core/vasio/VAsioConnection.hpp index daaf2a6c4..2f7c3e128 100644 --- a/SilKit/source/core/vasio/VAsioConnection.hpp +++ b/SilKit/source/core/vasio/VAsioConnection.hpp @@ -509,6 +509,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. + 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..381bcbeef 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -29,6 +29,50 @@ using namespace std::chrono_literals; +namespace { + +//! Maximum number of receive blobs held per thread, and the largest one worth keeping. +constexpr size_t ReceiveBlobPoolSize{8}; +constexpr size_t ReceiveBlobPoolMaxCapacity{64 * 1024}; + +/*! \brief Take a receive blob from the thread local pool, or allocate one. + * + * Every received message is linearised into a blob that deserialized payloads then alias, so the + * blob has to outlive the dispatch. Reusing one is therefore only safe once nothing references it + * any more, which is exactly what a use count of one means: only the pool still holds it. Blobs + * that no payload kept alive are reused without allocating, which is the common case because + * dispatch is synchronous. + */ +auto AcquireReceiveBlob(size_t size) -> std::shared_ptr> +{ + thread_local std::vector>> pool; + + // Large messages are rare; do not hoard a buffer for them. + if (size > ReceiveBlobPoolMaxCapacity) + { + return std::make_shared>(size); + } + + for (auto& entry : pool) + { + if (entry.use_count() == 1) + { + entry->resize(size); + return entry; + } + } + + auto blob = std::make_shared>(size); + if (pool.size() < ReceiveBlobPoolSize) + { + pool.push_back(blob); + } + + return blob; +} + +} // namespace + namespace SilKit { namespace Core { @@ -164,6 +208,9 @@ auto VAsioPeer::MakeSendItem(std::vector blob) -> SendItem { item.inlineSize = blob.size(); std::memcpy(item.inlineData.data(), blob.data(), item.inlineSize); + // NB: the bytes live in the item now, so hand the buffer back for the next serialization + // instead of letting it be freed. + RecycleSerializationBuffer(std::move(blob)); } else { @@ -369,8 +416,12 @@ void VAsioPeer::DispatchBuffer() } else { - // NB: linearised into a shared blob, so that deserialized payloads can alias it. - auto currentMsg = std::make_shared>(_currentMsgSize); + // NB: the message must be linearised out of the ring buffer because it may wrap, but + // it is allocated as a shared blob so that deserialized payloads can alias it + // instead of being copied out again. The blob is filled before being wrapped, + // which establishes the immutability the SharedSpan invariant requires. One blob + // per message keeps the retained memory bounded by the message's own size. + auto currentMsg = AcquireReceiveBlob(_currentMsgSize); if (!_msgBuffer.Read(SilKit::Util::ToSpan(*currentMsg))) { throw SilKitError("Reading data from ring buffer failed."); From c3be22c89eb334431ae37e0e1e4c6941ca549b5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Wed, 16 Sep 2026 10:34:03 +0200 Subject: [PATCH 3/6] vasio: avoid an allocation and a zero fill per message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent findings from profiling the small message path with callgrind and DHAT, 400,000 messages of 10 bytes between two participants. asio copies a buffer sequence into the async operation, which DHAT attributed as one allocation of 16 bytes per message. A single buffer is stored in the operation by value instead, so AsyncWriteSome and AsyncReadSome now take that path when the sequence holds exactly one buffer. That is the common case: a small message is written as one contiguous buffer, and the receive ring only needs a second buffer when its free region wraps. AcquireReceiveBlob resized a pooled blob to the exact message size, value initializing bytes that RingBuffer::Read overwrites immediately afterwards. Pooled blobs now only grow and stay at their high water size, and the caller reads and views exactly the message's bytes. This part is a regression from the commit that introduced the pool. Signed-off-by: Marius Börschig --- SilKit/source/core/vasio/VAsioPeer.cpp | 18 +++++-- .../io/impl/AsioGenericRawByteStream.cpp | 52 +++++++++++++------ 2 files changed, 50 insertions(+), 20 deletions(-) diff --git a/SilKit/source/core/vasio/VAsioPeer.cpp b/SilKit/source/core/vasio/VAsioPeer.cpp index 381bcbeef..f5e218f74 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -57,7 +57,13 @@ auto AcquireReceiveBlob(size_t size) -> std::shared_ptr> { if (entry.use_count() == 1) { - entry->resize(size); + // NB: only grow. Shrinking to the exact size would value initialize bytes that the + // caller overwrites immediately anyway, so keep the blob at its high water size + // and let the caller view only the part it filled. + if (entry->size() < size) + { + entry->resize(size); + } return entry; } } @@ -421,13 +427,15 @@ void VAsioPeer::DispatchBuffer() // instead of being copied out again. The blob is filled before being wrapped, // which establishes the immutability the SharedSpan invariant requires. One blob // per message keeps the retained memory bounded by the message's own size. - auto currentMsg = AcquireReceiveBlob(_currentMsgSize); - if (!_msgBuffer.Read(SilKit::Util::ToSpan(*currentMsg))) + const size_t blobSize = _currentMsgSize; + auto currentMsg = AcquireReceiveBlob(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)}; diff --git a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp index 0a1e481f6..cce54f417 100644 --- a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp +++ b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp @@ -119,14 +119,24 @@ 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: same as in AsyncWriteSome. The ring buffer only needs two buffers when the free + // region wraps, so a single buffer is the common case here too. + 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 +163,26 @@ 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 copies a buffer sequence into the async operation, which costs an allocation + // per write. A single buffer is stored in the operation by value instead, so take that + // path when there is only one. That is the common case, because a small message is + // written as one contiguous buffer. + 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); }); + } } } From ac46a556957470e187908f98334d70041469c3a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Thu, 24 Sep 2026 14:50:15 +0200 Subject: [PATCH 4/6] vasio: make receive buffer pool configurable and cover shared span paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - replace thread_local receive blob pool with ReceiveBlobPool owned by VAsioConnection - add Experimental.UseReceiveBufferPool (default true) to disable pooling for diagnosis - reuse best-fitting blob, cap slack at 4 KiB, acquire fence after use_count check - add VAsioPeer tests: inline/owned/shared send items, partial writes, aggregation, receive paths - add ReceiveBlobPool, VAsioProxyPeer and serdes aliasing / RPC wire format tests - add fan-out ITests: pubsub 1:3 (aggregation, pool off), RPC 1:2, large Ethernet frames, mixed direct/proxied subscribers - use non-uniform payloads in pubsub ITest harness - document UseReceiveBufferPool Signed-off-by: Marius Börschig --- .../ITest_Internals_DataPubSub.cpp | 62 ++- .../ITest_Internals_DataPubSub.hpp | 45 +- .../IntegrationTests/ITest_Internals_Rpc.cpp | 33 ++ .../ITest_RequestRemoteParticipantConnect.cpp | 67 +++ .../ITest_ThreeEthController.cpp | 31 +- .../config/ParticipantConfiguration.cpp | 3 +- .../config/ParticipantConfiguration.hpp | 4 + .../ParticipantConfiguration.schema.json | 6 + .../ParticipantConfigurationFromXImpl.cpp | 5 + .../config/ParticipantConfiguration_Full.json | 3 +- .../config/ParticipantConfiguration_Full.yaml | 3 +- SilKit/source/config/Test_YamlParser.cpp | 25 + SilKit/source/config/YamlReader.cpp | 1 + SilKit/source/config/YamlValidator.cpp | 1 + SilKit/source/config/YamlWriter.cpp | 1 + SilKit/source/core/vasio/CMakeLists.txt | 6 + SilKit/source/core/vasio/ReceiveBlobPool.cpp | 91 +++ SilKit/source/core/vasio/ReceiveBlobPool.hpp | 53 ++ .../core/vasio/Test_ReceiveBlobPool.cpp | 171 ++++++ .../core/vasio/Test_SerializedMessage.cpp | 188 +++++++ .../core/vasio/Test_VAsioConnection.cpp | 24 + SilKit/source/core/vasio/Test_VAsioPeer.cpp | 526 ++++++++++++++++++ .../source/core/vasio/Test_VAsioProxyPeer.cpp | 98 ++++ SilKit/source/core/vasio/VAsioConnection.cpp | 12 +- SilKit/source/core/vasio/VAsioConnection.hpp | 3 + SilKit/source/core/vasio/VAsioPeer.cpp | 56 +- SilKit/source/core/vasio/VAsioPeer.hpp | 6 +- .../experimental-configuration.rst | 28 + 28 files changed, 1473 insertions(+), 79 deletions(-) create mode 100644 SilKit/source/core/vasio/ReceiveBlobPool.cpp create mode 100644 SilKit/source/core/vasio/ReceiveBlobPool.hpp create mode 100644 SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp create mode 100644 SilKit/source/core/vasio/Test_VAsioPeer.cpp create mode 100644 SilKit/source/core/vasio/Test_VAsioProxyPeer.cpp 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/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..1f44aebb8 --- /dev/null +++ b/SilKit/source/core/vasio/ReceiveBlobPool.hpp @@ -0,0 +1,53 @@ +// 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 the blob of their message, so a blob can only be reused once nothing + * references it any more, which is what a use count of one means: only the pool still holds it. + * Blobs that no payload kept alive are reused without allocating, which is the common case because + * dispatch is synchronous. + * + * A disabled pool allocates a fresh blob per message. A payload used after its handler returned + * then reads freed memory, which sanitizers and debug heaps detect, instead of silently reading the + * bytes of a later 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/Test_ReceiveBlobPool.cpp b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp new file mode 100644 index 000000000..7808f5fee --- /dev/null +++ b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp @@ -0,0 +1,171 @@ +// 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, reports_whether_it_is_enabled) +{ + EXPECT_TRUE(ReceiveBlobPool{true}.IsEnabled()); + EXPECT_FALSE(ReceiveBlobPool{false}.IsEnabled()); +} + +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..c367f3919 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,186 @@ 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_TRUE(payload.HasOwner()); + 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..7c06c3353 --- /dev/null +++ b/SilKit/source/core/vasio/Test_VAsioPeer.cpp @@ -0,0 +1,526 @@ +// 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 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; +} + +auto ToVector(SilKit::Util::Span span) -> std::vector +{ + return {span.begin(), span.end()}; +} + +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, 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, 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(ToVector(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], ToVector(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], ToVector(first.data.AsSpan())); + EXPECT_EQ(payloads[1], ToVector(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(ToVector(retained[0].data.AsSpan()), ToVector(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], ToVector(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], ToVector(large.data.AsSpan())); + EXPECT_EQ(payloads[1], ToVector(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], ToVector(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 604790998..c91d6bb08 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(), @@ -2035,12 +2043,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 2f7c3e128..f46e75a7f 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}; diff --git a/SilKit/source/core/vasio/VAsioPeer.cpp b/SilKit/source/core/vasio/VAsioPeer.cpp index f5e218f74..3bfaf4ec4 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -29,66 +29,18 @@ using namespace std::chrono_literals; -namespace { - -//! Maximum number of receive blobs held per thread, and the largest one worth keeping. -constexpr size_t ReceiveBlobPoolSize{8}; -constexpr size_t ReceiveBlobPoolMaxCapacity{64 * 1024}; - -/*! \brief Take a receive blob from the thread local pool, or allocate one. - * - * Every received message is linearised into a blob that deserialized payloads then alias, so the - * blob has to outlive the dispatch. Reusing one is therefore only safe once nothing references it - * any more, which is exactly what a use count of one means: only the pool still holds it. Blobs - * that no payload kept alive are reused without allocating, which is the common case because - * dispatch is synchronous. - */ -auto AcquireReceiveBlob(size_t size) -> std::shared_ptr> -{ - thread_local std::vector>> pool; - - // Large messages are rare; do not hoard a buffer for them. - if (size > ReceiveBlobPoolMaxCapacity) - { - return std::make_shared>(size); - } - - for (auto& entry : pool) - { - if (entry.use_count() == 1) - { - // NB: only grow. Shrinking to the exact size would value initialize bytes that the - // caller overwrites immediately anyway, so keep the blob at its high water size - // and let the caller view only the part it filled. - if (entry->size() < size) - { - entry->resize(size); - } - return entry; - } - } - - auto blob = std::make_shared>(size); - if (pool.size() < ReceiveBlobPoolSize) - { - pool.push_back(blob); - } - - return blob; -} - -} // namespace - 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); @@ -428,7 +380,7 @@ void VAsioPeer::DispatchBuffer() // which establishes the immutability the SharedSpan invariant requires. One blob // per message keeps the retained memory bounded by the message's own size. const size_t blobSize = _currentMsgSize; - auto currentMsg = AcquireReceiveBlob(blobSize); + 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. diff --git a/SilKit/source/core/vasio/VAsioPeer.hpp b/SilKit/source/core/vasio/VAsioPeer.hpp index e80a3fe4b..cf98d2ea6 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" @@ -49,8 +50,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; @@ -142,6 +145,7 @@ class VAsioPeer std::atomic _currentMsgSize{0u}; RingBuffer _msgBuffer; std::vector _currentReceivingBuffers; + ReceiveBlobPool* _receiveBlobPool{nullptr}; // sending mutable std::mutex _sendingQueueMutex; 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 From 2f8b749d4916b1122d993d77e7656c2155c96c6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Fri, 25 Sep 2026 08:08:24 +0200 Subject: [PATCH 5/6] remove dead code, trim comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marius Börschig --- SilKit/source/core/internal/MessageBuffer.hpp | 10 ++------ SilKit/source/core/vasio/ReceiveBlobPool.hpp | 11 +++------ SilKit/source/core/vasio/SilKitLink.hpp | 4 +--- .../core/vasio/Test_ReceiveBlobPool.cpp | 6 ----- .../core/vasio/Test_SerializedMessage.cpp | 1 - SilKit/source/core/vasio/Test_VAsioPeer.cpp | 24 ++++++++----------- SilKit/source/core/vasio/VAsioConnection.cpp | 12 +--------- SilKit/source/core/vasio/VAsioConnection.hpp | 2 +- SilKit/source/core/vasio/VAsioPeer.cpp | 8 +------ .../io/impl/AsioGenericRawByteStream.cpp | 9 +++---- 10 files changed, 22 insertions(+), 65 deletions(-) diff --git a/SilKit/source/core/internal/MessageBuffer.hpp b/SilKit/source/core/internal/MessageBuffer.hpp index d79666896..96c154806 100644 --- a/SilKit/source/core/internal/MessageBuffer.hpp +++ b/SilKit/source/core/internal/MessageBuffer.hpp @@ -62,10 +62,7 @@ inline auto SerializationBufferPool() -> std::vector>& /*! \brief Take a serialization buffer from the thread local free list, or a fresh one. * - * Serializing a message otherwise allocates a buffer every time and retires it again almost - * immediately, which heap profiling showed as one allocation per sent message. The list is thread - * local so it needs no synchronization; a buffer acquired on one thread and recycled on another - * simply migrates between lists, which is harmless. + * A buffer acquired on one thread and recycled on another migrates between lists, which is harmless. */ inline auto AcquireSerializationBuffer() -> std::vector { @@ -77,7 +74,6 @@ inline auto AcquireSerializationBuffer() -> std::vector auto buffer = std::move(pool.back()); pool.pop_back(); - buffer.clear(); // keeps the capacity, which is the point of pooling it return buffer; } @@ -105,9 +101,7 @@ class MessageBuffer public: // ---------------------------------------- // Constructors and Destructor - //! NB: takes its storage from the thread local pool, so serializing a message need not - //! allocate one. Only this writing constructor does so; the reading constructors below - //! bring their own storage. + // NB: the writing constructor takes its storage from the thread local serialization pool. inline MessageBuffer() : _storage{AcquireSerializationBuffer()} { diff --git a/SilKit/source/core/vasio/ReceiveBlobPool.hpp b/SilKit/source/core/vasio/ReceiveBlobPool.hpp index 1f44aebb8..8d3ba33c1 100644 --- a/SilKit/source/core/vasio/ReceiveBlobPool.hpp +++ b/SilKit/source/core/vasio/ReceiveBlobPool.hpp @@ -14,14 +14,9 @@ namespace Core { /*! \brief The blobs that received messages are linearised into. * - * Deserialized payloads alias the blob of their message, so a blob can only be reused once nothing - * references it any more, which is what a use count of one means: only the pool still holds it. - * Blobs that no payload kept alive are reused without allocating, which is the common case because - * dispatch is synchronous. - * - * A disabled pool allocates a fresh blob per message. A payload used after its handler returned - * then reads freed memory, which sanitizers and debug heaps detect, instead of silently reading the - * bytes of a later message. See Config::Experimental::useReceiveBufferPool. + * 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. */ diff --git a/SilKit/source/core/vasio/SilKitLink.hpp b/SilKit/source/core/vasio/SilKitLink.hpp index d110a47a6..5808deb8c 100644 --- a/SilKit/source/core/vasio/SilKitLink.hpp +++ b/SilKit/source/core/vasio/SilKitLink.hpp @@ -72,9 +72,7 @@ class SilKitLink Services::Logging::ILoggerInternal* _logger; Services::Orchestration::ITimeProvider* _timeProvider; - //! A local receiver together with its service endpoint. The endpoint is resolved once when the - //! receiver is registered, because resolving it per message required a dynamic_cast, and these - //! receivers use multiple inheritance so that cast has to walk the inheritance graph. + //! The service endpoint is resolved on registration, to avoid a dynamic_cast per message. struct LocalReceiver { ReceiverT* receiver; diff --git a/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp index 7808f5fee..272be789d 100644 --- a/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp +++ b/SilKit/source/core/vasio/Test_ReceiveBlobPool.cpp @@ -21,12 +21,6 @@ using WeakBlob = std::weak_ptr>; constexpr size_t Small{100}; constexpr size_t Large{60 * 1024}; -TEST(Test_ReceiveBlobPool, reports_whether_it_is_enabled) -{ - EXPECT_TRUE(ReceiveBlobPool{true}.IsEnabled()); - EXPECT_FALSE(ReceiveBlobPool{false}.IsEnabled()); -} - TEST(Test_ReceiveBlobPool, reuses_an_unreferenced_blob) { ReceiveBlobPool pool{true}; diff --git a/SilKit/source/core/vasio/Test_SerializedMessage.cpp b/SilKit/source/core/vasio/Test_SerializedMessage.cpp index c367f3919..27cdaa4d2 100644 --- a/SilKit/source/core/vasio/Test_SerializedMessage.cpp +++ b/SilKit/source/core/vasio/Test_SerializedMessage.cpp @@ -265,7 +265,6 @@ void ExpectPayloadAliasesTheBlob(const MsgT& msg, PayloadOf payloadOf) const auto& payload = payloadOf(received); const auto view = payload.AsSpan(); - EXPECT_TRUE(payload.HasOwner()); 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())); diff --git a/SilKit/source/core/vasio/Test_VAsioPeer.cpp b/SilKit/source/core/vasio/Test_VAsioPeer.cpp index 7c06c3353..6349ef3c7 100644 --- a/SilKit/source/core/vasio/Test_VAsioPeer.cpp +++ b/SilKit/source/core/vasio/Test_VAsioPeer.cpp @@ -28,6 +28,7 @@ namespace { using namespace SilKit::Core; using SilKit::Services::PubSub::WireDataMessageEvent; +using SilKit::Util::ToStdVector; using testing::_; using testing::NiceMock; @@ -65,11 +66,6 @@ auto Concat(std::initializer_list> parts) -> std::vector span) -> std::vector -{ - return {span.begin(), span.end()}; -} - struct TestPeerListener : IVAsioPeerListener { std::function onData; @@ -418,7 +414,7 @@ class Test_VAsioPeerReceive : public testing::TestWithParam p.Listener().onData = [this](SerializedMessage& message) { storageSizes.push_back(message.GetStorageSize()); auto event = message.Deserialize(); - payloads.push_back(ToVector(event.data.AsSpan())); + payloads.push_back(ToStdVector(event.data.AsSpan())); if (retain) { retained.push_back(std::move(event)); @@ -441,7 +437,7 @@ TEST_P(Test_VAsioPeerReceive, payload_is_delivered_intact) p.Feed(Reference(event, 3)); ASSERT_EQ(payloads.size(), 1u); - EXPECT_EQ(payloads[0], ToVector(event.data.AsSpan())); + EXPECT_EQ(payloads[0], ToStdVector(event.data.AsSpan())); } TEST_P(Test_VAsioPeerReceive, message_wrapping_the_ring_buffer_is_reassembled) @@ -457,8 +453,8 @@ TEST_P(Test_VAsioPeerReceive, message_wrapping_the_ring_buffer_is_reassembled) p.Feed(Concat({firstBytes, Reference(second, 3)}), {4010}); ASSERT_EQ(payloads.size(), 2u); - EXPECT_EQ(payloads[0], ToVector(first.data.AsSpan())); - EXPECT_EQ(payloads[1], ToVector(second.data.AsSpan())); + 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) @@ -476,7 +472,7 @@ TEST_P(Test_VAsioPeerReceive, retained_payload_is_not_overwritten) } ASSERT_EQ(retained.size(), 1u); - EXPECT_EQ(ToVector(retained[0].data.AsSpan()), ToVector(kept.data.AsSpan())); + EXPECT_EQ(ToStdVector(retained[0].data.AsSpan()), ToStdVector(kept.data.AsSpan())); } TEST_P(Test_VAsioPeerReceive, message_after_a_larger_one_is_viewed_exactly) @@ -491,7 +487,7 @@ TEST_P(Test_VAsioPeerReceive, message_after_a_larger_one_is_viewed_exactly) ASSERT_EQ(payloads.size(), 2u); EXPECT_EQ(storageSizes[1], smallBytes.size()); - EXPECT_EQ(payloads[1], ToVector(small.data.AsSpan())); + EXPECT_EQ(payloads[1], ToStdVector(small.data.AsSpan())); } TEST_P(Test_VAsioPeerReceive, message_above_the_pool_limit) @@ -503,8 +499,8 @@ TEST_P(Test_VAsioPeerReceive, message_above_the_pool_limit) p.Feed(Reference(small, 3)); ASSERT_EQ(payloads.size(), 2u); - EXPECT_EQ(payloads[0], ToVector(large.data.AsSpan())); - EXPECT_EQ(payloads[1], ToVector(small.data.AsSpan())); + EXPECT_EQ(payloads[0], ToStdVector(large.data.AsSpan())); + EXPECT_EQ(payloads[1], ToStdVector(small.data.AsSpan())); } TEST_P(Test_VAsioPeerReceive, byte_by_byte_delivery) @@ -515,7 +511,7 @@ TEST_P(Test_VAsioPeerReceive, byte_by_byte_delivery) p.Feed(bytes, std::deque(bytes.size(), 1)); ASSERT_EQ(payloads.size(), 1u); - EXPECT_EQ(payloads[0], ToVector(event.data.AsSpan())); + EXPECT_EQ(payloads[0], ToStdVector(event.data.AsSpan())); } INSTANTIATE_TEST_SUITE_P(ReceiveBlobPool, Test_VAsioPeerReceive, testing::Values(true, false), diff --git a/SilKit/source/core/vasio/VAsioConnection.cpp b/SilKit/source/core/vasio/VAsioConnection.cpp index c91d6bb08..1b61bfcd8 100644 --- a/SilKit/source/core/vasio/VAsioConnection.cpp +++ b/SilKit/source/core/vasio/VAsioConnection.cpp @@ -1791,17 +1791,7 @@ void VAsioConnection::ReceiveRawSilKitMessage(IVAsioPeer* from, SerializedMessag auto endpoint = buffer.GetEndpointAddress(); //ExtractEndpointAddress(buffer); - // NB: IVAsioPeer derives publicly from IServiceEndpoint, so this is a plain upcast. It used - // to be a dynamic_cast, which profiling showed to be a measurable share of the work done - // for every received message. - const IServiceEndpoint* fromService = from; - - // NB: assign into a reusable descriptor rather than constructing a fresh one. The copy exists - // only to override the service id, and constructing it allocated the nodes of the - // supplemental data map for every received message. Assignment reuses the existing nodes - // and string buffers. Only the io thread reaches this and the dispatch below is - // synchronous, so one scratch instance suffices. - _receiveServiceDescriptor = fromService->GetServiceDescriptor(); + _receiveServiceDescriptor = from->GetServiceDescriptor(); _receiveServiceDescriptor.SetServiceId(endpoint.endpoint); _vasioReceivers[receiverIdx]->ReceiveRawMsg(from, _receiveServiceDescriptor, std::move(buffer)); diff --git a/SilKit/source/core/vasio/VAsioConnection.hpp b/SilKit/source/core/vasio/VAsioConnection.hpp index f46e75a7f..5906d1d72 100644 --- a/SilKit/source/core/vasio/VAsioConnection.hpp +++ b/SilKit/source/core/vasio/VAsioConnection.hpp @@ -513,7 +513,7 @@ class VAsioConnection 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. + //! supplemental data map each time. Only touched on the io thread, and dispatch is synchronous. ServiceDescriptor _receiveServiceDescriptor; std::vector> _vasioReceivers; diff --git a/SilKit/source/core/vasio/VAsioPeer.cpp b/SilKit/source/core/vasio/VAsioPeer.cpp index 3bfaf4ec4..8d19bbdd1 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -166,8 +166,6 @@ auto VAsioPeer::MakeSendItem(std::vector blob) -> SendItem { item.inlineSize = blob.size(); std::memcpy(item.inlineData.data(), blob.data(), item.inlineSize); - // NB: the bytes live in the item now, so hand the buffer back for the next serialization - // instead of letting it be freed. RecycleSerializationBuffer(std::move(blob)); } else @@ -374,11 +372,7 @@ void VAsioPeer::DispatchBuffer() } else { - // NB: the message must be linearised out of the ring buffer because it may wrap, but - // it is allocated as a shared blob so that deserialized payloads can alias it - // instead of being copied out again. The blob is filled before being wrapped, - // which establishes the immutability the SharedSpan invariant requires. One blob - // per message keeps the retained memory bounded by the message's own size. + // NB: linearised into a shared blob, so that deserialized payloads can alias it. const size_t blobSize = _currentMsgSize; auto currentMsg = _receiveBlobPool->Acquire(blobSize); diff --git a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp index cce54f417..2afc2c20c 100644 --- a/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp +++ b/SilKit/source/core/vasio/io/impl/AsioGenericRawByteStream.cpp @@ -119,8 +119,7 @@ void AsioGenericRawByteStream::AsyncReadSome(MutableBufferSequence bufferSequenc _reading = true; } - // NB: same as in AsyncWriteSome. The ring buffer only needs two buffers when the free - // region wraps, so a single buffer is the common case here too. + // 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()}, @@ -163,10 +162,8 @@ void AsioGenericRawByteStream::AsyncWriteSome(ConstBufferSequence bufferSequence _writing = true; } - // NB: asio copies a buffer sequence into the async operation, which costs an allocation - // per write. A single buffer is stored in the operation by value instead, so take that - // path when there is only one. That is the common case, because a small message is - // written as one contiguous buffer. + // 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()}, From 20bedb2420ca9cc4ed9ae25df9c54ddb98b944e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marius=20B=C3=B6rschig?= Date: Fri, 25 Sep 2026 13:17:03 +0200 Subject: [PATCH 6/6] fix aggregation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Marius Börschig --- Demos/tools/Benchmark/performance-tests.toml | 21 +--- SilKit/source/core/vasio/Test_VAsioPeer.cpp | 110 +++++++++++++++++++ SilKit/source/core/vasio/VAsioPeer.cpp | 52 ++++++--- SilKit/source/core/vasio/VAsioPeer.hpp | 12 +- 4 files changed, 157 insertions(+), 38 deletions(-) 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/source/core/vasio/Test_VAsioPeer.cpp b/SilKit/source/core/vasio/Test_VAsioPeer.cpp index 6349ef3c7..89aaf5c51 100644 --- a/SilKit/source/core/vasio/Test_VAsioPeer.cpp +++ b/SilKit/source/core/vasio/Test_VAsioPeer.cpp @@ -347,6 +347,99 @@ TEST(Test_VAsioPeer, queued_items_keep_valid_headers) 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; @@ -384,6 +477,23 @@ TEST(Test_VAsioPeer, aggregation_concatenates_inline_owned_and_shared_items) 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; diff --git a/SilKit/source/core/vasio/VAsioPeer.cpp b/SilKit/source/core/vasio/VAsioPeer.cpp index 8d19bbdd1..62f4922d4 100644 --- a/SilKit/source/core/vasio/VAsioPeer.cpp +++ b/SilKit/source/core/vasio/VAsioPeer.cpp @@ -154,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)); } } @@ -194,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(body.data(), body.size()); + _currentSendingBuffers.emplace_back(item.ownedBody.data(), item.ownedBody.size()); + } + else + { + const auto body = item.sharedBody.AsSpan(); + if (!body.empty()) + { + _currentSendingBuffers.emplace_back(body.data(), body.size()); + } } } } @@ -279,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(); @@ -441,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 cf98d2ea6..07d085cb5 100644 --- a/SilKit/source/core/vasio/VAsioPeer.hpp +++ b/SilKit/source/core/vasio/VAsioPeer.hpp @@ -40,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 @@ -106,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(); @@ -150,9 +156,9 @@ class VAsioPeer // 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;