From 163c72e69ae67eef46217ac289f7b8a77ff4864d Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Fri, 9 Oct 2026 13:25:03 +0200 Subject: [PATCH 1/6] TcpByteTransport: receive small chunks with one recv ReceiveChunkAsync read the 8-byte chunk header and then the body, two receive calls per chunk. On loopback, receive syscalls are a large share of the client CPU per small request. The transport now waits for data with a small staging buffer (8 KB, allocated on the first receive, not from the BufferManager budget). One receive returns the header and, when it is small and already there, the whole chunk; bytes of following chunks are kept for the next call. A chunk larger than the staging buffer is completed with a receive into the rented buffer, as before. The receive buffer is still rented only after the header has been validated, so an idle peer holds no budget; message type and size checks are unchanged. A cancelled receive keeps the bytes that already arrived. Co-Authored-By: Claude Opus 5.5 --- src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs | 162 +++++++++++++----- .../Stack/Transport/TcpByteTransportTests.cs | 116 ++++++++++++- 2 files changed, 235 insertions(+), 43 deletions(-) diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs index e7eb5d927a..88c55240c1 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs @@ -332,21 +332,26 @@ public async ValueTask> ReceiveChunkAsync(CancellationToken c { Socket socket = RequireConnectedSocket(); int receiveBufferSize = Volatile.Read(ref m_receiveBufferSize); + if (Interlocked.Exchange(ref m_receiving, 1) != 0) + { + throw new InvalidOperationException("A receive is already in progress."); + } - // Read the 8-byte UASC chunk header (message type + chunk size) before - // renting the receive buffer: an idle or silent peer must not hold a - // full receive buffer (and its share of a process-wide memory budget) - // while the transport waits for the next chunk. - byte[] header = Interlocked.Exchange(ref m_receiveHeader, null) ?? - new byte[TcpMessageLimits.MessageTypeAndSize]; - int messageSize; - byte[] buffer; try { - await ReadExactAsync(socket, header, 0, TcpMessageLimits.MessageTypeAndSize, ct) - .ConfigureAwait(false); + // An idle peer is waited for with the small staging buffer only: the + // receive buffer (and its share of a process-wide memory budget) is + // rented once the chunk header has arrived and been validated. One + // receive into the staging buffer returns the whole chunk when it is + // small and already there, plus any bytes of the following chunks, + // which are kept for the next call. + byte[] staging = m_staging ??= AllocateStaging(); + while (m_stagedCount < TcpMessageLimits.MessageTypeAndSize) + { + await FillStagingAsync(socket, staging, ct).ConfigureAwait(false); + } - uint messageType = BitConverter.ToUInt32(header, 0); + uint messageType = BitConverter.ToUInt32(staging, m_stagedOffset); if (!TcpMessageType.IsValid(messageType)) { throw ServiceResultException.Create( @@ -355,7 +360,7 @@ await ReadExactAsync(socket, header, 0, TcpMessageLimits.MessageTypeAndSize, ct) messageType); } - messageSize = BitConverter.ToInt32(header, 4); + int messageSize = BitConverter.ToInt32(staging, m_stagedOffset + 4); if (messageSize <= TcpMessageLimits.MessageTypeAndSize || messageSize > receiveBufferSize) { @@ -366,37 +371,55 @@ await ReadExactAsync(socket, header, 0, TcpMessageLimits.MessageTypeAndSize, ct) receiveBufferSize); } + // A chunk that fits the staging buffer is completed there; the rest + // of a larger one is read directly into the rented buffer. + if (messageSize <= staging.Length) + { + while (m_stagedCount < messageSize) + { + await FillStagingAsync(socket, staging, ct).ConfigureAwait(false); + } + } + // Rent for this chunk rather than for the largest one the // channel accepts (as PipeByteTransport does): a small chunk, // in particular one kept for an incomplete message, must not // hold a whole receive buffer of memory budget. - buffer = m_bufferManager.TakeBuffer( + byte[] buffer = m_bufferManager.TakeBuffer( messageSize, nameof(ReceiveChunkAsync), ct); - Buffer.BlockCopy(header, 0, buffer, 0, TcpMessageLimits.MessageTypeAndSize); - } - finally - { - Volatile.Write(ref m_receiveHeader, header); - } - - try - { - m_bufferManager.Lock(buffer); try { - // Read the remaining bytes (chunk body) directly after the header. - await ReadExactAsync( - socket, - buffer, - TcpMessageLimits.MessageTypeAndSize, - messageSize - TcpMessageLimits.MessageTypeAndSize, - ct) - .ConfigureAwait(false); + m_bufferManager.Lock(buffer); + try + { + int staged = Math.Min(m_stagedCount, messageSize); + Buffer.BlockCopy(staging, m_stagedOffset, buffer, 0, staged); + m_stagedOffset += staged; + m_stagedCount -= staged; + if (m_stagedCount == 0) + { + m_stagedOffset = 0; + } + + if (staged < messageSize) + { + await ReadExactAsync( + socket, + buffer, + staged, + messageSize - staged, + ct) + .ConfigureAwait(false); + } + } + finally + { + m_bufferManager.Unlock(buffer); + } var segment = new ArraySegment(buffer, 0, messageSize); - m_bufferManager.Unlock(buffer); buffer = null!; // ownership transferred to caller return segment; } @@ -404,21 +427,67 @@ await ReadExactAsync( { if (buffer != null) { - m_bufferManager.Unlock(buffer); + m_bufferManager.ReturnBuffer(buffer, nameof(ReceiveChunkAsync)); } throw; } } - catch + finally { - if (buffer != null) - { - m_bufferManager.ReturnBuffer(buffer, nameof(ReceiveChunkAsync)); - } - throw; + Volatile.Write(ref m_receiving, 0); } } + /// + /// Receives whatever is available into the free space of the staging buffer, + /// after moving the staged bytes to its start. + /// +#if NET6_0_OR_GREATER + [AsyncMethodBuilder(typeof(PoolingAsyncValueTaskMethodBuilder))] +#endif + private async ValueTask FillStagingAsync(Socket socket, byte[] staging, CancellationToken ct) + { + if (m_stagedOffset > 0) + { + Buffer.BlockCopy(staging, m_stagedOffset, staging, 0, m_stagedCount); + m_stagedOffset = 0; + } + + int n; +#if NET5_0_OR_GREATER + n = await socket + .ReceiveAsync( + new Memory(staging, m_stagedCount, staging.Length - m_stagedCount), + SocketFlags.None, + ct) + .ConfigureAwait(false); +#else + n = await socket + .ReceiveAsync( + new ArraySegment(staging, m_stagedCount, staging.Length - m_stagedCount), + SocketFlags.None) + .ConfigureAwait(false); + ct.ThrowIfCancellationRequested(); +#endif + if (n <= 0) + { + throw ServiceResultException.Create( + StatusCodes.BadConnectionClosed, + "Remote side closed the connection."); + } + m_stagedCount += n; + } + + private static byte[] AllocateStaging() + { +#if NET5_0_OR_GREATER + // A pending receive pins the buffer for as long as the peer is idle. + return GC.AllocateUninitializedArray(kStagingSize, pinned: true); +#else + return new byte[kStagingSize]; +#endif + } + void IUaSCByteTransportLimits.SetReceiveBufferSize(int receiveBufferSize) { if (receiveBufferSize <= TcpMessageLimits.MessageTypeAndSize) @@ -625,9 +694,20 @@ private void ShutdownAndDispose(Socket socket) private int m_receiveBufferSize; /// - /// Reusable chunk-header buffer; taken for the duration of a header read. + /// The size of the staging buffer that receives chunk headers, small chunks + /// and the bytes of following chunks that arrived with them. + /// + private const int kStagingSize = 8192; + + /// + /// Staging buffer, allocated on the first receive. The bytes from + /// on ( of them) + /// were received but not returned yet. /// - private byte[]? m_receiveHeader; + private byte[]? m_staging; + private int m_stagedOffset; + private int m_stagedCount; + private int m_receiving; private readonly ILogger m_logger; [System.Diagnostics.CodeAnalysis.SuppressMessage( diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs index f01e8e5c11..2699ee2d50 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs @@ -29,6 +29,7 @@ using System; using System.Collections.Generic; +using System.Linq; using System.Net; using System.Net.Sockets; using System.Reflection; @@ -553,6 +554,116 @@ await serverSocket } } + [Test] + public async Task ReceiveChunkAsyncReturnsChunksSentTogetherOneAtATimeAsync() + { + const int largeBufferSize = 65536; + m_bufferManager = new BufferManager("TcpByteTransportTests", largeBufferSize, m_telemetry); + (TcpByteTransport client, Socket serverSocket, TcpListener listener) = + await CreateConnectedPairAsync(largeBufferSize).ConfigureAwait(false); + using var _l = new ListenerScope(listener); + using Socket _s = serverSocket; + using (client) + { + // Small chunks around one larger than the staging buffer, in one send: + // each receive returns exactly one chunk and keeps the rest. + byte[][] chunks = + [ + BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 40), + BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 9), + BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 20000), + BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 8192), + BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 100) + ]; + byte[] all = [.. chunks.SelectMany(c => c)]; + await serverSocket + .SendAsync(new ArraySegment(all), SocketFlags.None) + .ConfigureAwait(false); + + foreach (byte[] expected in chunks) + { + await AssertReceivesAsync(client, expected).ConfigureAwait(false); + } + } + } + + [Test] + public async Task ReceiveChunkAsyncCompletesAChunkSplitAcrossSendsAsync() + { + (TcpByteTransport client, Socket serverSocket, TcpListener listener) = + await CreateConnectedPairAsync().ConfigureAwait(false); + using var _l = new ListenerScope(listener); + using Socket _s = serverSocket; + using (client) + { + // The first chunk arrives in three pieces, the middle one ending inside + // the header of the second chunk. + byte[] first = BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 64); + byte[] second = BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 32); + byte[] all = [.. first, .. second]; + Task> receive = client.ReceiveChunkAsync(CancellationToken.None).AsTask(); + foreach ((int offset, int count) in new[] { (0, 3), (3, 64), (67, 29) }) + { + await serverSocket + .SendAsync(new ArraySegment(all, offset, count), SocketFlags.None) + .ConfigureAwait(false); + await Task.Delay(50).ConfigureAwait(false); + } + + AssertChunk(await receive.ConfigureAwait(false), first); + await AssertReceivesAsync(client, second).ConfigureAwait(false); + } + } + + [Test] + public async Task ReceiveChunkAsyncHonoursCancellationAndKeepsStagedBytesAsync() + { + (TcpByteTransport client, Socket serverSocket, TcpListener listener) = + await CreateConnectedPairAsync().ConfigureAwait(false); + using var _l = new ListenerScope(listener); + using Socket _s = serverSocket; + using (client) + { + // Half a header, then the receive is cancelled. + byte[] chunk = BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 48); + await serverSocket + .SendAsync(new ArraySegment(chunk, 0, 4), SocketFlags.None) + .ConfigureAwait(false); + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(200)); + Assert.That( + async () => await client.ReceiveChunkAsync(cts.Token).ConfigureAwait(false), + Throws.InstanceOf()); + + // The next receive continues with the bytes that already arrived. + await serverSocket + .SendAsync(new ArraySegment(chunk, 4, chunk.Length - 4), SocketFlags.None) + .ConfigureAwait(false); + await AssertReceivesAsync(client, chunk).ConfigureAwait(false); + } + } + + private async Task AssertReceivesAsync(TcpByteTransport client, byte[] expected) + { + ArraySegment chunk = await client + .ReceiveChunkAsync(CancellationToken.None) + .AsTask() + .WaitAsync(TimeSpan.FromSeconds(5)) + .ConfigureAwait(false); + AssertChunk(chunk, expected); + } + + private void AssertChunk(ArraySegment chunk, byte[] expected) + { + try + { + Assert.That(chunk.ToArray(), Is.EqualTo(expected)); + } + finally + { + m_bufferManager.ReturnBuffer(chunk.Array, nameof(AssertChunk)); + } + } + [Test] public async Task ReceiveChunkAsyncDoesNotHoldReceiveBufferWhileIdle() { @@ -704,7 +815,8 @@ public void SendChunkAsyncOnDisconnectedTransportFails() Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadConnectionClosed)); } - private async Task<(TcpByteTransport, Socket, TcpListener)> CreateConnectedPairAsync() + private async Task<(TcpByteTransport, Socket, TcpListener)> CreateConnectedPairAsync( + int receiveBufferSize = kBufferSize) { var listener = new TcpListener(IPAddress.Loopback, 0); listener.Start(); @@ -712,7 +824,7 @@ public void SendChunkAsyncOnDisconnectedTransportFails() Task acceptTask = listener.AcceptSocketAsync(); - var transport = new TcpByteTransport(m_bufferManager, kBufferSize, m_telemetry); + var transport = new TcpByteTransport(m_bufferManager, receiveBufferSize, m_telemetry); await transport .ConnectAsync(new Uri($"opc.tcp://127.0.0.1:{port}"), CancellationToken.None) .ConfigureAwait(false); From 2dcf8cdb26cd72903f63e86205d1423cbba2e236 Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Fri, 9 Oct 2026 13:25:04 +0200 Subject: [PATCH 2/6] ManagedSession: take the service reader lock without a second state machine Every service call was an async wrapper around the inner session call: await the reader lock, then await the call, one more async state machine box per request on top of the inner session's. When services are available and no writer holds or waits for the lock, the reader lock is now taken synchronously (AsyncReaderWriterLock TryEnterReader) and released by a cached continuation on the inner call's task, registered before the caller's await, so the lock is free again when the caller resumes (unless the call completes while the continuation is being registered; the release then runs on the pool). Otherwise the call waits for the lock as before, so reconnect and failover still block new calls and wait for the calls in flight. A synchronous exception from the inner call is still reported through the returned task. Co-Authored-By: Claude Opus 5.5 --- .../Session/ManagedSession.Services.cs | 570 ++++++++++-------- .../Reconnect/ConnectionStateMachine.cs | 6 + .../Utils/AsyncReaderWriterLock.cs | 71 +++ .../Utils/AsyncReaderWriterLockTests.cs | 86 +++ 4 files changed, 466 insertions(+), 267 deletions(-) diff --git a/src/Opc.Ua.Client/Session/ManagedSession.Services.cs b/src/Opc.Ua.Client/Session/ManagedSession.Services.cs index 7c0366b7a1..05a8846c1a 100644 --- a/src/Opc.Ua.Client/Session/ManagedSession.Services.cs +++ b/src/Opc.Ua.Client/Session/ManagedSession.Services.cs @@ -27,6 +27,8 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ +using System; +using System.Runtime.ExceptionServices; using System.Threading; using System.Threading.Tasks; @@ -41,22 +43,22 @@ namespace Opc.Ua.Client public partial class ManagedSession { /// - public async ValueTask ReadAsync( + public ValueTask ReadAsync( RequestHeader? requestHeader, double maxAge, TimestampsToReturn timestampsToReturn, ArrayOf nodesToRead, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.ReadAsync( - requestHeader, maxAge, timestampsToReturn, nodesToRead, ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, maxAge, timestampsToReturn, nodesToRead), + static (session, a, ct) => session.ReadAsync( + a.requestHeader, a.maxAge, a.timestampsToReturn, a.nodesToRead, ct), + ct); } /// - public async ValueTask HistoryReadAsync( + public ValueTask HistoryReadAsync( RequestHeader? requestHeader, ExtensionObject historyReadDetails, TimestampsToReturn timestampsToReturn, @@ -64,199 +66,177 @@ public async ValueTask HistoryReadAsync( ArrayOf nodesToRead, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.HistoryReadAsync( - requestHeader, historyReadDetails, timestampsToReturn, - releaseContinuationPoints, nodesToRead, ct).ConfigureAwait(false); - } + return CallServiceAsync( + ( + requestHeader, + historyReadDetails, + timestampsToReturn, + releaseContinuationPoints, + nodesToRead), + static (session, a, ct) => session.HistoryReadAsync( + a.requestHeader, + a.historyReadDetails, + a.timestampsToReturn, + a.releaseContinuationPoints, + a.nodesToRead, + ct), + ct); } /// - public async ValueTask WriteAsync( + public ValueTask WriteAsync( RequestHeader? requestHeader, ArrayOf nodesToWrite, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.WriteAsync( - requestHeader, - nodesToWrite, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, nodesToWrite), + static (session, a, ct) => session.WriteAsync( + a.requestHeader, a.nodesToWrite, ct), + ct); } /// - public async ValueTask HistoryUpdateAsync( + public ValueTask HistoryUpdateAsync( RequestHeader? requestHeader, ArrayOf historyUpdateDetails, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.HistoryUpdateAsync( - requestHeader, - historyUpdateDetails, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, historyUpdateDetails), + static (session, a, ct) => session.HistoryUpdateAsync( + a.requestHeader, a.historyUpdateDetails, ct), + ct); } /// - public async ValueTask BrowseAsync( + public ValueTask BrowseAsync( RequestHeader? requestHeader, ViewDescription? view, uint requestedMaxReferencesPerNode, ArrayOf nodesToBrowse, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.BrowseAsync( - requestHeader, - view, - requestedMaxReferencesPerNode, - nodesToBrowse, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, view, requestedMaxReferencesPerNode, nodesToBrowse), + static (session, a, ct) => session.BrowseAsync( + a.requestHeader, a.view, a.requestedMaxReferencesPerNode, a.nodesToBrowse, ct), + ct); } /// - public async ValueTask BrowseNextAsync( + public ValueTask BrowseNextAsync( RequestHeader? requestHeader, bool releaseContinuationPoints, ArrayOf continuationPoints, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.BrowseNextAsync( - requestHeader, - releaseContinuationPoints, - continuationPoints, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, releaseContinuationPoints, continuationPoints), + static (session, a, ct) => session.BrowseNextAsync( + a.requestHeader, a.releaseContinuationPoints, a.continuationPoints, ct), + ct); } /// - public async ValueTask TranslateBrowsePathsToNodeIdsAsync( + public ValueTask TranslateBrowsePathsToNodeIdsAsync( RequestHeader? requestHeader, ArrayOf browsePaths, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.TranslateBrowsePathsToNodeIdsAsync( - requestHeader, - browsePaths, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, browsePaths), + static (session, a, ct) => session.TranslateBrowsePathsToNodeIdsAsync( + a.requestHeader, a.browsePaths, ct), + ct); } /// - public async ValueTask RegisterNodesAsync( + public ValueTask RegisterNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToRegister, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.RegisterNodesAsync( - requestHeader, - nodesToRegister, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, nodesToRegister), + static (session, a, ct) => session.RegisterNodesAsync( + a.requestHeader, a.nodesToRegister, ct), + ct); } /// - public async ValueTask UnregisterNodesAsync( + public ValueTask UnregisterNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToUnregister, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.UnregisterNodesAsync( - requestHeader, - nodesToUnregister, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, nodesToUnregister), + static (session, a, ct) => session.UnregisterNodesAsync( + a.requestHeader, a.nodesToUnregister, ct), + ct); } /// - public async ValueTask CallAsync( + public ValueTask CallAsync( RequestHeader? requestHeader, ArrayOf methodsToCall, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.CallAsync( - requestHeader, - methodsToCall, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, methodsToCall), + static (session, a, ct) => session.CallAsync( + a.requestHeader, a.methodsToCall, ct), + ct); } /// - public async ValueTask CreateMonitoredItemsAsync( + public ValueTask CreateMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToCreate, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.CreateMonitoredItemsAsync( - requestHeader, - subscriptionId, - timestampsToReturn, - itemsToCreate, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, timestampsToReturn, itemsToCreate), + static (session, a, ct) => session.CreateMonitoredItemsAsync( + a.requestHeader, a.subscriptionId, a.timestampsToReturn, a.itemsToCreate, ct), + ct); } /// - public async ValueTask ModifyMonitoredItemsAsync( + public ValueTask ModifyMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToModify, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.ModifyMonitoredItemsAsync( - requestHeader, - subscriptionId, - timestampsToReturn, - itemsToModify, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, timestampsToReturn, itemsToModify), + static (session, a, ct) => session.ModifyMonitoredItemsAsync( + a.requestHeader, a.subscriptionId, a.timestampsToReturn, a.itemsToModify, ct), + ct); } /// - public async ValueTask SetMonitoringModeAsync( + public ValueTask SetMonitoringModeAsync( RequestHeader? requestHeader, uint subscriptionId, MonitoringMode monitoringMode, ArrayOf monitoredItemIds, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.SetMonitoringModeAsync( - requestHeader, - subscriptionId, - monitoringMode, - monitoredItemIds, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, monitoringMode, monitoredItemIds), + static (session, a, ct) => session.SetMonitoringModeAsync( + a.requestHeader, a.subscriptionId, a.monitoringMode, a.monitoredItemIds, ct), + ct); } /// - public async ValueTask SetTriggeringAsync( + public ValueTask SetTriggeringAsync( RequestHeader? requestHeader, uint subscriptionId, uint triggeringItemId, @@ -264,37 +244,34 @@ public async ValueTask SetTriggeringAsync( ArrayOf linksToRemove, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.SetTriggeringAsync( - requestHeader, - subscriptionId, - triggeringItemId, - linksToAdd, - linksToRemove, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, triggeringItemId, linksToAdd, linksToRemove), + static (session, a, ct) => session.SetTriggeringAsync( + a.requestHeader, + a.subscriptionId, + a.triggeringItemId, + a.linksToAdd, + a.linksToRemove, + ct), + ct); } /// - public async ValueTask DeleteMonitoredItemsAsync( + public ValueTask DeleteMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, ArrayOf monitoredItemIds, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.DeleteMonitoredItemsAsync( - requestHeader, - subscriptionId, - monitoredItemIds, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, monitoredItemIds), + static (session, a, ct) => session.DeleteMonitoredItemsAsync( + a.requestHeader, a.subscriptionId, a.monitoredItemIds, ct), + ct); } /// - public async ValueTask CreateSubscriptionAsync( + public ValueTask CreateSubscriptionAsync( RequestHeader? requestHeader, double requestedPublishingInterval, uint requestedLifetimeCount, @@ -304,22 +281,29 @@ public async ValueTask CreateSubscriptionAsync( byte priority, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.CreateSubscriptionAsync( + return CallServiceAsync( + ( requestHeader, requestedPublishingInterval, requestedLifetimeCount, requestedMaxKeepAliveCount, maxNotificationsPerPublish, publishingEnabled, - priority, - ct).ConfigureAwait(false); - } + priority), + static (session, a, ct) => session.CreateSubscriptionAsync( + a.requestHeader, + a.requestedPublishingInterval, + a.requestedLifetimeCount, + a.requestedMaxKeepAliveCount, + a.maxNotificationsPerPublish, + a.publishingEnabled, + a.priority, + ct), + ct); } /// - public async ValueTask ModifySubscriptionAsync( + public ValueTask ModifySubscriptionAsync( RequestHeader? requestHeader, uint subscriptionId, double requestedPublishingInterval, @@ -329,163 +313,149 @@ public async ValueTask ModifySubscriptionAsync( byte priority, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.ModifySubscriptionAsync( + return CallServiceAsync( + ( requestHeader, subscriptionId, requestedPublishingInterval, requestedLifetimeCount, requestedMaxKeepAliveCount, maxNotificationsPerPublish, - priority, - ct).ConfigureAwait(false); - } + priority), + static (session, a, ct) => session.ModifySubscriptionAsync( + a.requestHeader, + a.subscriptionId, + a.requestedPublishingInterval, + a.requestedLifetimeCount, + a.requestedMaxKeepAliveCount, + a.maxNotificationsPerPublish, + a.priority, + ct), + ct); } /// - public async ValueTask SetPublishingModeAsync( + public ValueTask SetPublishingModeAsync( RequestHeader? requestHeader, bool publishingEnabled, ArrayOf subscriptionIds, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.SetPublishingModeAsync( - requestHeader, - publishingEnabled, - subscriptionIds, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, publishingEnabled, subscriptionIds), + static (session, a, ct) => session.SetPublishingModeAsync( + a.requestHeader, a.publishingEnabled, a.subscriptionIds, ct), + ct); } /// - public async ValueTask PublishAsync( + public ValueTask PublishAsync( RequestHeader? requestHeader, ArrayOf subscriptionAcknowledgements, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.PublishAsync( - requestHeader, - subscriptionAcknowledgements, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionAcknowledgements), + static (session, a, ct) => session.PublishAsync( + a.requestHeader, a.subscriptionAcknowledgements, ct), + ct); } /// - public async ValueTask RepublishAsync( + public ValueTask RepublishAsync( RequestHeader? requestHeader, uint subscriptionId, uint retransmitSequenceNumber, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.RepublishAsync( - requestHeader, - subscriptionId, - retransmitSequenceNumber, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionId, retransmitSequenceNumber), + static (session, a, ct) => session.RepublishAsync( + a.requestHeader, a.subscriptionId, a.retransmitSequenceNumber, ct), + ct); } /// - public async ValueTask TransferSubscriptionsAsync( + public ValueTask TransferSubscriptionsAsync( RequestHeader? requestHeader, ArrayOf subscriptionIds, bool sendInitialValues, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.TransferSubscriptionsAsync( - requestHeader, - subscriptionIds, - sendInitialValues, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionIds, sendInitialValues), + static (session, a, ct) => session.TransferSubscriptionsAsync( + a.requestHeader, a.subscriptionIds, a.sendInitialValues, ct), + ct); } /// - public async ValueTask DeleteSubscriptionsAsync( + public ValueTask DeleteSubscriptionsAsync( RequestHeader? requestHeader, ArrayOf subscriptionIds, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.DeleteSubscriptionsAsync( - requestHeader, - subscriptionIds, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, subscriptionIds), + static (session, a, ct) => session.DeleteSubscriptionsAsync( + a.requestHeader, a.subscriptionIds, ct), + ct); } /// - public async ValueTask AddNodesAsync( + public ValueTask AddNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToAdd, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.AddNodesAsync( - requestHeader, - nodesToAdd, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, nodesToAdd), + static (session, a, ct) => session.AddNodesAsync( + a.requestHeader, a.nodesToAdd, ct), + ct); } /// - public async ValueTask AddReferencesAsync( + public ValueTask AddReferencesAsync( RequestHeader? requestHeader, ArrayOf referencesToAdd, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.AddReferencesAsync( - requestHeader, - referencesToAdd, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, referencesToAdd), + static (session, a, ct) => session.AddReferencesAsync( + a.requestHeader, a.referencesToAdd, ct), + ct); } /// - public async ValueTask DeleteNodesAsync( + public ValueTask DeleteNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToDelete, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.DeleteNodesAsync( - requestHeader, - nodesToDelete, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, nodesToDelete), + static (session, a, ct) => session.DeleteNodesAsync( + a.requestHeader, a.nodesToDelete, ct), + ct); } /// - public async ValueTask DeleteReferencesAsync( + public ValueTask DeleteReferencesAsync( RequestHeader? requestHeader, ArrayOf referencesToDelete, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.DeleteReferencesAsync( - requestHeader, - referencesToDelete, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, referencesToDelete), + static (session, a, ct) => session.DeleteReferencesAsync( + a.requestHeader, a.referencesToDelete, ct), + ct); } /// - public async ValueTask QueryFirstAsync( + public ValueTask QueryFirstAsync( RequestHeader? requestHeader, ViewDescription? view, ArrayOf nodeTypes, @@ -494,38 +464,35 @@ public async ValueTask QueryFirstAsync( uint maxReferencesToReturn, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.QueryFirstAsync( - requestHeader, - view, - nodeTypes, - filter, - maxDataSetsToReturn, - maxReferencesToReturn, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, view, nodeTypes, filter, maxDataSetsToReturn, maxReferencesToReturn), + static (session, a, ct) => session.QueryFirstAsync( + a.requestHeader, + a.view, + a.nodeTypes, + a.filter, + a.maxDataSetsToReturn, + a.maxReferencesToReturn, + ct), + ct); } /// - public async ValueTask QueryNextAsync( + public ValueTask QueryNextAsync( RequestHeader? requestHeader, bool releaseContinuationPoint, ByteString continuationPoint, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.QueryNextAsync( - requestHeader, - releaseContinuationPoint, - continuationPoint, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, releaseContinuationPoint, continuationPoint), + static (session, a, ct) => session.QueryNextAsync( + a.requestHeader, a.releaseContinuationPoint, a.continuationPoint, ct), + ct); } /// - public async ValueTask CreateSessionAsync( + public ValueTask CreateSessionAsync( RequestHeader? requestHeader, ApplicationDescription? clientDescription, string? serverUri, @@ -537,9 +504,8 @@ public async ValueTask CreateSessionAsync( uint maxResponseMessageSize, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.CreateSessionAsync( + return CallServiceAsync( + ( requestHeader, clientDescription, serverUri, @@ -548,13 +514,23 @@ public async ValueTask CreateSessionAsync( clientNonce, clientCertificate, requestedSessionTimeout, - maxResponseMessageSize, - ct).ConfigureAwait(false); - } + maxResponseMessageSize), + static (session, a, ct) => session.CreateSessionAsync( + a.requestHeader, + a.clientDescription, + a.serverUri, + a.endpointUrl, + a.sessionName, + a.clientNonce, + a.clientCertificate, + a.requestedSessionTimeout, + a.maxResponseMessageSize, + ct), + ct); } /// - public async ValueTask ActivateSessionAsync( + public ValueTask ActivateSessionAsync( RequestHeader? requestHeader, SignatureData? clientSignature, ArrayOf clientSoftwareCertificates, @@ -563,47 +539,107 @@ public async ValueTask ActivateSessionAsync( SignatureData? userTokenSignature, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.ActivateSessionAsync( + return CallServiceAsync( + ( requestHeader, clientSignature, clientSoftwareCertificates, localeIds, userIdentityToken, - userTokenSignature, - ct).ConfigureAwait(false); - } + userTokenSignature), + static (session, a, ct) => session.ActivateSessionAsync( + a.requestHeader, + a.clientSignature, + a.clientSoftwareCertificates, + a.localeIds, + a.userIdentityToken, + a.userTokenSignature, + ct), + ct); } /// - public async ValueTask CloseSessionAsync( + public ValueTask CloseSessionAsync( RequestHeader? requestHeader, bool deleteSubscriptions, CancellationToken ct) { - using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) - { - return await InnerSession.CloseSessionAsync( - requestHeader, - deleteSubscriptions, - ct).ConfigureAwait(false); - } + return CallServiceAsync( + (requestHeader, deleteSubscriptions), + static (session, a, ct) => session.CloseSessionAsync( + a.requestHeader, a.deleteSubscriptions, ct), + ct); } /// - public async ValueTask CancelAsync( + public ValueTask CancelAsync( RequestHeader? requestHeader, uint requestHandle, CancellationToken ct) + { + return CallServiceAsync( + (requestHeader, requestHandle), + static (session, a, ct) => session.CancelAsync( + a.requestHeader, a.requestHandle, ct), + ct); + } + + /// + /// Calls a service on the inner session under the reader lock. + /// + /// + /// When services are available and no writer holds or waits for the + /// lock, the reader lock is taken synchronously and released when the + /// call completes, before the caller's continuation runs. That avoids + /// a second async state machine per call; otherwise the call waits + /// for the lock as does. + /// + /// The service arguments, passed to the static call. + /// The service response. + private ValueTask CallServiceAsync( + TArgs args, + Func> call, + CancellationToken ct) + { + if ((StateMachine.AreServicesAvailable || StateMachine.IsWorkerFlow) && + m_serviceLock.TryEnterReader()) + { + ValueTask operation; + try + { + operation = call(InnerSession, args, ct); + } + catch (Exception ex) + { + m_serviceLock.ExitReader(); + return RethrowAsync(ExceptionDispatchInfo.Capture(ex)); + } + return m_serviceLock.ExitReaderOnCompletion(operation); + } + return CallServiceLockedAsync(args, call, ct); + } + + private async ValueTask CallServiceLockedAsync( + TArgs args, + Func> call, + CancellationToken ct) { using (await m_serviceLock.ReaderLockAsync(ct).ConfigureAwait(false)) { - return await InnerSession.CancelAsync( - requestHeader, - requestHandle, - ct).ConfigureAwait(false); + return await call(InnerSession, args, ct).ConfigureAwait(false); } } + + /// + /// Reports an exception thrown synchronously by a call the way the async + /// path does: a cancellation as a canceled task, anything else as a fault. + /// + /// The service response. + private static async ValueTask RethrowAsync(ExceptionDispatchInfo error) + { + await Task.CompletedTask.ConfigureAwait(false); + error.Throw(); + return default!; + } } } diff --git a/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs b/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs index 5fd231723a..2cad6b93d0 100644 --- a/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs +++ b/src/Opc.Ua.Client/Session/Reconnect/ConnectionStateMachine.cs @@ -235,6 +235,12 @@ private static TimeSpan NormalizeMaxTotalReconnectTime(TimeSpan? maxTotalReconne /// internal bool IsWorkerFlow => m_inWorkerFlow.Value is { Active: true }; + /// + /// Whether service calls may proceed now, the condition + /// waits for. + /// + internal bool AreServicesAvailable => Volatile.Read(ref m_servicesAvailable); + /// /// Event raised when the state changes. /// diff --git a/src/Opc.Ua.Client/Utils/AsyncReaderWriterLock.cs b/src/Opc.Ua.Client/Utils/AsyncReaderWriterLock.cs index 69c5ae9325..b4395d0b59 100644 --- a/src/Opc.Ua.Client/Utils/AsyncReaderWriterLock.cs +++ b/src/Opc.Ua.Client/Utils/AsyncReaderWriterLock.cs @@ -91,6 +91,14 @@ namespace Opc.Ua.Client /// internal sealed class AsyncReaderWriterLock : IDisposable { + /// + /// Creates the lock. + /// + public AsyncReaderWriterLock() + { + m_releaseReader = ReleaseReader; + } + /// /// Optional asynchronous operation that completes before a reader /// acquires the lock. @@ -138,6 +146,68 @@ public async ValueTask ReaderLockAsync( return new Releaser(this, isWriter: false); } + /// + /// Acquires a reader lock without waiting when no writer holds or + /// waits for the lock. Does not run ; + /// the caller checks the precondition it stands for. Release with + /// or . + /// + /// True if the reader lock was acquired. + public bool TryEnterReader() + { + if (!m_writer.Wait(0)) + { + return false; + } + try + { + lock (m_state) + { + m_activeReaders++; + } + } + finally + { + m_writer.Release(); + } + return true; + } + + /// + /// Releases a reader lock acquired with . + /// + public void ExitReader() + { + ReleaseReader(); + } + + /// + /// Releases a reader lock acquired with + /// when completes, and returns a value + /// task for the same result. + /// + /// + /// The release is the first continuation registered on the operation's + /// task and runs synchronously when it completes, so the lock is normally + /// free again by the time an awaiter of the returned task resumes, as with + /// a using block around the awaited operation. Only if the task + /// completes while the continuation is being registered does the release + /// run on the thread pool instead, shortly after. The delegate is cached, + /// so this allocates nothing beyond the task's continuation list. + /// + /// The result of the operation. + public ValueTask ExitReaderOnCompletion(ValueTask operation) + { + if (operation.IsCompleted) + { + ReleaseReader(); + return operation; + } + Task task = operation.AsTask(); + task.ConfigureAwait(false).GetAwaiter().UnsafeOnCompleted(m_releaseReader); + return new ValueTask(task); + } + /// /// Asynchronously acquires the writer lock. Excludes both /// other writers and all readers for the duration the lock @@ -320,6 +390,7 @@ public void Dispose() } } + private readonly Action m_releaseReader; private bool m_isDisposed; private readonly SemaphoreSlim m_writer = new(1, 1); private readonly Lock m_state = new(); diff --git a/tests/Opc.Ua.Client.Tests/Utils/AsyncReaderWriterLockTests.cs b/tests/Opc.Ua.Client.Tests/Utils/AsyncReaderWriterLockTests.cs index 73a6d4f406..fc57d11d17 100644 --- a/tests/Opc.Ua.Client.Tests/Utils/AsyncReaderWriterLockTests.cs +++ b/tests/Opc.Ua.Client.Tests/Utils/AsyncReaderWriterLockTests.cs @@ -427,5 +427,91 @@ public async Task ManyParallelReadersAdmittedAsync(CancellationToken ct) Assert.That(peakHolding, Is.EqualTo(kReaders), "All 32 readers should have held the lock simultaneously."); } + + [Test] + public async Task TryEnterReaderFailsWhileAWriterHoldsTheLockAsync() + { + using var rwLock = new AsyncReaderWriterLock(); + using (await rwLock.WriterLockAsync().ConfigureAwait(false)) + { + Assert.That(rwLock.TryEnterReader(), Is.False); + } + Assert.That(rwLock.TryEnterReader(), Is.True); + rwLock.ExitReader(); + } + + [Test] + public async Task ReaderFromTryEnterIsHeldUntilTheOperationCompletesAsync() + { + using var rwLock = new AsyncReaderWriterLock(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var operation = new TaskCompletionSource(); + Assert.That(rwLock.TryEnterReader(), Is.True); + ValueTask call = rwLock.ExitReaderOnCompletion(new ValueTask(operation.Task)); + + Task writer = rwLock.WriterLockAsync(timeout.Token).AsTask(); + await Task.Delay(50, timeout.Token).ConfigureAwait(false); + Assert.That(writer.IsCompleted, Is.False, "the writer waits for the reader to drain"); + + operation.SetResult(42); + using (await writer.WaitAsync(timeout.Token).ConfigureAwait(false)) + { + Assert.That(await call.ConfigureAwait(false), Is.EqualTo(42)); + } + } + + [Test] + public async Task AwaiterResumesAfterTheReaderFromTryEnterIsReleasedAsync() + { + using var rwLock = new AsyncReaderWriterLock(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var operation = new TaskCompletionSource(); + Assert.That(rwLock.TryEnterReader(), Is.True); + ValueTask call = rwLock.ExitReaderOnCompletion(new ValueTask(operation.Task)); + + // A writer requested in the awaiter's continuation finds no reader left, + // as after a using block around the awaited operation. + bool writerAcquiredAtOnce = false; + Task resumed = Task.Run(async () => + { + await call.ConfigureAwait(false); + ValueTask writer = rwLock.WriterLockAsync(timeout.Token); + writerAcquiredAtOnce = writer.IsCompleted; + (await writer.ConfigureAwait(false)).Dispose(); + }); + await Task.Delay(50, timeout.Token).ConfigureAwait(false); + operation.SetResult(42); + + await resumed.WaitAsync(timeout.Token).ConfigureAwait(false); + Assert.That(writerAcquiredAtOnce, Is.True); + } + + [Test] + public async Task ReaderFromTryEnterIsReleasedWhenTheOperationFailsAsync() + { + using var rwLock = new AsyncReaderWriterLock(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + var operation = new TaskCompletionSource(); + Assert.That(rwLock.TryEnterReader(), Is.True); + ValueTask call = rwLock.ExitReaderOnCompletion(new ValueTask(operation.Task)); + operation.SetException(new InvalidOperationException("failed")); + + Assert.That(async () => await call.ConfigureAwait(false), Throws.InvalidOperationException); + using AsyncReaderWriterLock.Releaser writer = + await rwLock.WriterLockAsync(timeout.Token).ConfigureAwait(false); + } + + [Test] + public async Task ReaderFromTryEnterIsReleasedAtOnceForACompletedOperationAsync() + { + using var rwLock = new AsyncReaderWriterLock(); + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + Assert.That(rwLock.TryEnterReader(), Is.True); + ValueTask call = rwLock.ExitReaderOnCompletion(new ValueTask(7)); + Assert.That(call.IsCompleted, Is.True); + Assert.That(await call.ConfigureAwait(false), Is.EqualTo(7)); + using AsyncReaderWriterLock.Releaser writer = + await rwLock.WriterLockAsync(timeout.Token).ConfigureAwait(false); + } } } From 005593f05f0735c928968700a2fb25dfeaa9c194 Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Fri, 9 Oct 2026 13:25:23 +0200 Subject: [PATCH 3/6] perf: --client shared runs sessions on ClientChannelManager channels Co-Authored-By: Claude Opus 5.5 --- perf/README.md | 2 +- perf/ServerLoadHarness/Program.cs | 30 ++++++++++++++++++++++-------- perf/ab.sh | 3 ++- 3 files changed, 25 insertions(+), 10 deletions(-) diff --git a/perf/README.md b/perf/README.md index b77fc87615..273c01d742 100644 --- a/perf/README.md +++ b/perf/README.md @@ -37,7 +37,7 @@ perf/ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe run --scenario | `--nodes` | 100 | Values per Read/Write request, or monitored items per subscription. | | `--inflight` | 1 | Outstanding requests per session (closed loop). | | `--security` | `none` | `none`, `sign` or `encrypt` (Basic256Sha256). | -| `--client` | `classic` | `classic`: `Session` with the classic subscription engine. `managed`: `ManagedSession` with the V2 subscription engine (`DefaultSubscriptionEngineFactory`). | +| `--client` | `classic` | `classic`: `Session` with the classic subscription engine. `managed`: `ManagedSession` with the V2 subscription engine (`DefaultSubscriptionEngineFactory`). `shared`: classic sessions on secure channels shared through one `ClientChannelManager`, created one at a time (`--connect-parallel`). | | `--callback` | `fast` | Classic data changes through `Subscription.FastDataChangeCallback` (`fast`) or the per-item `MonitoredItem.Notification` event (`item`). | | `--subs`, `--pub`, `--write-interval` | 5, 100, 100 | Subscriptions per session, publishing interval and writer interval in ms (`sub`). | | `--duration`, `--warmup` | 20, 10 | Measurement window and warm-up in seconds. | diff --git a/perf/ServerLoadHarness/Program.cs b/perf/ServerLoadHarness/Program.cs index d449fa97f3..83879fc8de 100644 --- a/perf/ServerLoadHarness/Program.cs +++ b/perf/ServerLoadHarness/Program.cs @@ -47,6 +47,7 @@ using System.Globalization; using System.IO; using System.Linq; +using System.Runtime.CompilerServices; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -301,25 +302,38 @@ await client // classic: Session from the default session factory (classic subscription engine). // managed: ManagedSession with the default V2 subscription engine. + // shared: classic Sessions whose secure channels come from one ClientChannelManager. string clientKind = Arg("client", "classic"); - ConfiguredEndpoint? managedEndpoint = clientKind == "managed" + ConfiguredEndpoint? endpointForBuilder = clientKind is "managed" or "shared" ? await client.GetEndpointAsync(new Uri(url), policy, selected).ConfigureAwait(false) : null; + var channelManager = new ClientChannelManager(client.Config, telemetry); + await using ConfiguredAsyncDisposable disposeChannelManager = channelManager.ConfigureAwait(false); var sessions = new ISession[sessionCount]; var connectTime = Stopwatch.StartNew(); await Parallel.ForAsync( 0, sessionCount, - new ParallelOptions { MaxDegreeOfParallelism = 32 }, - async (i, ct) => sessions[i] = managedEndpoint != null - ? await new ManagedSessionBuilder(client.Config, telemetry) - .UseEndpoint(managedEndpoint) + new ParallelOptions { MaxDegreeOfParallelism = clientKind == "shared" ? ArgInt("connect-parallel", 1) : 32 }, + async (i, ct) => sessions[i] = clientKind switch + { + "managed" => await new ManagedSessionBuilder(client.Config, telemetry) + .UseEndpoint(endpointForBuilder!) .WithSessionTimeout(TimeSpan.FromSeconds(60)) .ConnectAsync(ct) - .ConfigureAwait(false) - : await client + .ConfigureAwait(false), + "shared" => await Session.CreateAsync( + channelManager, + client.Config, + endpointForBuilder!, + updateBeforeConnect: false, + sessionName: "PerfShared", + ct: ct) + .ConfigureAwait(false), + _ => await client .ConnectAsync(new Uri(url), policy, selected) - .ConfigureAwait(false)) + .ConfigureAwait(false) + }) .ConfigureAwait(false); Console.WriteLine(FormattableString.Invariant( $"connected {sessionCount} {clientKind} sessions ({security}) in {connectTime.ElapsedMilliseconds} ms")); diff --git a/perf/ab.sh b/perf/ab.sh index aa8c4e5984..bcecfb4753 100644 --- a/perf/ab.sh +++ b/perf/ab.sh @@ -1,10 +1,11 @@ #!/bin/bash # Paired A/B runs: alternates a baseline and a candidate harness build per scenario and round. # BASELINE: harness built in a second worktree of the commit to compare against. +# CANDIDATE: harness to compare (default: the build in this worktree). # SCENARIOS: file with one scenario per line (default: server-scenarios.txt). # ROUNDS: rounds per scenario (default 3). OUT: result file (default ab-results.txt). A=${BASELINE:-C:/uabase/perf/ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe} -B=./ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe +B=${CANDIDATE:-./ServerLoadHarness/bin/Release/net10.0/ServerLoadHarness.exe} O=${OUT:-ab-results.txt}; rm -f "$O" ROUNDS=${ROUNDS:-3} SCENARIOS=${SCENARIOS:-server-scenarios.txt} From 740ccec339de7f754842b4d6945cce8ba663401b Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Fri, 9 Oct 2026 13:25:23 +0200 Subject: [PATCH 4/6] UaSCBinaryChannel: coalesce queued writes into one vectored send The channel writer sent queued writes one by one, one socket send each. With several requests in flight on a channel (a shared client channel, or a server answering concurrent requests) writes queue up behind the one being sent. When more writes for the same transport are queued, the writer now takes up to 16 of them and sends their segments with one vectored send, if the transport carries a byte stream (TcpByteTransport, through the admission lease on the server, opts in with the internal IUaSCChunkCoalescing). Each write is then reported on its own, in order, with its size; a failed send is reported for every write in it, and their buffers are returned. Writes still never start on the caller's stack, and a single writer still sends one batch at a time, so chunk order is unchanged. Transports that frame each send (WebSocket, capture decorators) are not coalesced. Co-Authored-By: Claude Opus 5.5 --- .../Stack/Tcp/IUaSCByteTransport.cs | 15 +++ src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs | 6 + .../Stack/Tcp/UaSCBinaryChannel.cs | 125 +++++++++++++++++- .../Stack/Tcp/UaScConnectionAdmission.cs | 12 +- .../UaSCBinaryChannelWriteQueueTests.cs | 118 ++++++++++++++++- 5 files changed, 265 insertions(+), 11 deletions(-) diff --git a/src/Opc.Ua.Core/Stack/Tcp/IUaSCByteTransport.cs b/src/Opc.Ua.Core/Stack/Tcp/IUaSCByteTransport.cs index f4f9019ef1..b0d85d4cbc 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/IUaSCByteTransport.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/IUaSCByteTransport.cs @@ -158,6 +158,21 @@ internal interface IUaSCByteTransportLimits void SetReceiveBufferSize(int receiveBufferSize); } + /// + /// Implemented by byte transports that carry a plain byte stream, so that + /// several complete chunks can go out in one + /// . + /// Transports that frame each send (a WebSocket message, for instance) or + /// observe chunks one by one do not implement it. + /// + internal interface IUaSCChunkCoalescing + { + /// + /// Whether one send may carry several chunks back to back. + /// + bool CanCoalesceChunks { get; } + } + /// /// Factory used by client-side transport channels to create an /// unconnected instance configured with diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs index 88c55240c1..fe8709613f 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs @@ -84,6 +84,7 @@ public IUaSCByteTransport Create( public sealed class TcpByteTransport : IUaSCByteTransport, IUaSCByteTransportLimits, + IUaSCChunkCoalescing, IDisposable { /// @@ -488,6 +489,11 @@ private static byte[] AllocateStaging() #endif } + /// + /// A TCP stream carries chunks back to back. + /// + bool IUaSCChunkCoalescing.CanCoalesceChunks => true; + void IUaSCByteTransportLimits.SetReceiveBufferSize(int receiveBufferSize) { if (receiveBufferSize <= TcpMessageLimits.MessageTypeAndSize) diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs index 4a581f4ce5..dcaaab9ee3 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs @@ -30,6 +30,7 @@ using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; +using System.Runtime.InteropServices; using System.Text; using System.Threading; using System.Threading.Tasks; @@ -1491,6 +1492,7 @@ private void QueueWrite(PendingWrite write) /// private async Task RunWritesAsync() { + List batch = m_writeBatch; while (true) { PendingWrite write; @@ -1502,11 +1504,34 @@ private async Task RunWritesAsync() return; } write = m_writeQueue.Dequeue(); + + // Writes queued behind this one for the same transport go out + // with it in one vectored send, when the transport is a stream. + if (m_writeQueue.Count > 0 && + write.Transport is IUaSCChunkCoalescing coalescing && + coalescing.CanCoalesceChunks) + { + batch.Add(write); + while (m_writeQueue.Count > 0 && + batch.Count < kMaxCoalescedWrites && + ReferenceEquals(m_writeQueue.Peek().Transport, write.Transport)) + { + batch.Add(m_writeQueue.Dequeue()); + } + } } - Task pending = write.Buffers != null - ? WriteBuffersAsync(write.Transport, write.Buffers, write.State) - : WriteSingleChunkAsync(write.Transport, write.Chunk, write.BackingBuffer!, write.State); + Task pending; + if (batch.Count > 0) + { + pending = WriteCoalescedAsync(write.Transport, batch); + } + else + { + pending = write.Buffers != null + ? WriteBuffersAsync(write.Transport, write.Buffers, write.State) + : WriteSingleChunkAsync(write.Transport, write.Chunk, write.BackingBuffer!, write.State); + } if (pending.IsCompleted) { // Observe a fault that escaped the write's own error handling, so @@ -1666,6 +1691,88 @@ private async Task WriteSingleChunkAsync( } } + /// + /// Sends several queued writes in one vectored send, then reports each + /// of them in order: with its own size when the send succeeded, and with + /// the error for all of them when it failed, since the bytes that reached + /// the transport cannot be attributed to a write. + /// + /// The transport all writes in the batch target. + /// The writes, in queue order. Cleared on return. + private async Task WriteCoalescedAsync( + IUaSCByteTransport transport, + List batch) + { + BufferCollection gathered = m_writeBatchBuffers; + ServiceResult result = ServiceResult.Good; + try + { + foreach (PendingWrite write in batch) + { + if (write.Buffers != null) + { + gathered.AddRange(write.Buffers); + } + else + { + gathered.Add(GetSegment(write.Chunk)); + } + } + await transport.SendChunkAsync(gathered, CancellationToken.None).ConfigureAwait(false); + } + catch (ServiceResultException sre) + { + result = sre.Result; + } + catch (Exception ex) + { + result = ServiceResult.Create( + ex, + StatusCodes.BadTcpInternalError, + "Unexpected error during write operation."); + } + finally + { + gathered.Clear(); + bool good = ServiceResult.IsGood(result); + foreach (PendingWrite write in batch) + { + int sent; + if (write.Buffers != null) + { + sent = good ? write.Buffers.TotalSize : 0; + } + else + { + sent = good ? write.Chunk.Length : 0; + try + { + if (write.BackingBuffer != null) + { + BufferManager.ReturnBuffer(write.BackingBuffer, "WriteChunkAsync"); + } + } + catch + { + // Best-effort, as in WriteSingleChunkAsync. + } + } + ReportWriteComplete(write.Buffers, write.State, sent, result); + } + batch.Clear(); + } + + static ArraySegment GetSegment(ReadOnlyMemory chunk) + { + if (!MemoryMarshal.TryGetArray(chunk, out ArraySegment segment)) + { + // BeginWriteMessage always wraps an array. + throw new InvalidOperationException("The chunk does not wrap an array."); + } + return segment; + } + } + private async Task WriteBuffersAsync( IUaSCByteTransport transport, BufferCollection buffers, @@ -2168,6 +2275,18 @@ public void UpdateLastActiveTime() private readonly Lock m_writeQueueLock = new(); private readonly Queue m_writeQueue = new(); private bool m_writerRunning; + + /// + /// The most writes sent together in one vectored send. + /// + private const int kMaxCoalescedWrites = 16; + + /// + /// The writes and the gathered segments of the current coalesced send, + /// used only by the single running writer. + /// + private readonly List m_writeBatch = []; + private readonly BufferCollection m_writeBatchBuffers = []; private long m_lastActiveTimestamp; private readonly string m_contextId; private readonly ILogger m_logger; diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaScConnectionAdmission.cs b/src/Opc.Ua.Core/Stack/Tcp/UaScConnectionAdmission.cs index 421f4e6ca4..bfd2dcdd2e 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaScConnectionAdmission.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaScConnectionAdmission.cs @@ -249,7 +249,11 @@ public void Start() /// Follows a transport through reverse-connect handoff. Only physical close releases capacity. /// internal sealed class Lease : - IUaSCByteTransport, IUaSCByteTransportLimits, IUaSCHandshakeCompletionSource, IDisposable + IUaSCByteTransport, + IUaSCByteTransportLimits, + IUaSCChunkCoalescing, + IUaSCHandshakeCompletionSource, + IDisposable { /// /// Takes ownership of the acquired reservations and the original handshake start timestamp. @@ -485,6 +489,12 @@ void IUaSCByteTransportLimits.SetReceiveBufferSize(int receiveBufferSize) } } + /// + /// Whether the attached transport accepts several chunks in one send. + /// + bool IUaSCChunkCoalescing.CanCoalesceChunks => + m_transport is IUaSCChunkCoalescing coalescing && coalescing.CanCoalesceChunks; + /// /// Claims expiry against handshake completion using the original, non-renewable deadline. /// diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs index eaad377e35..7765e940f8 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs @@ -149,6 +149,92 @@ public async Task AFailedSendIsReportedAndLaterWritesStillGoOutAsync() Assert.That(channel.HasPendingWrites, Is.False); } + /// + /// On a stream transport, writes queued behind a send in progress go out + /// together in order, at most 16 per send, and each is reported on its own. + /// + [Test] + [CancelAfter(30000)] + public async Task QueuedWritesAreCoalescedIntoFewerSendsInOrderAsync() + { + var transport = new RecordingTransport { CanCoalesceChunks = true }; + using var channel = new WriteProbeChannel(m_telemetry, transport); + const int count = 40; + + channel.Write(0, useCollection: false); + await transport.FirstSendStarted.Task.ConfigureAwait(false); + for (int ii = 1; ii < count; ii++) + { + channel.Write(ii, useCollection: ii % 2 == 0); + } + transport.ReleaseFirstSend.SetResult(true); + + await channel.WaitForCompletionsAsync(count).ConfigureAwait(false); + + Assert.That(transport.Sent, Is.EqualTo(Range(count))); + Assert.That(transport.SendCalls, Is.EqualTo(4), "1 + 16 + 16 + 7"); + Assert.That(transport.MaxConcurrentSends, Is.EqualTo(1)); + Assert.That(channel.Completed, Is.EqualTo(Range(count))); + Assert.That(channel.BytesReported, Is.EqualTo(count * sizeof(int))); + Assert.That(channel.HasPendingWrites, Is.False); + } + + /// + /// A transport that frames each send keeps one write per send. + /// + [Test] + [CancelAfter(30000)] + public async Task WritesAreNotCoalescedForAFramingTransportAsync() + { + var transport = new RecordingTransport(); + using var channel = new WriteProbeChannel(m_telemetry, transport); + const int count = 10; + + channel.Write(0, useCollection: false); + await transport.FirstSendStarted.Task.ConfigureAwait(false); + for (int ii = 1; ii < count; ii++) + { + channel.Write(ii, useCollection: ii % 2 == 0); + } + transport.ReleaseFirstSend.SetResult(true); + + await channel.WaitForCompletionsAsync(count).ConfigureAwait(false); + + Assert.That(transport.Sent, Is.EqualTo(Range(count))); + Assert.That(transport.SendCalls, Is.EqualTo(count)); + } + + /// + /// A failed coalesced send is reported for every write in it; the writes + /// queued after it still go out. + /// + [Test] + [CancelAfter(30000)] + public async Task AFailedCoalescedSendIsReportedForEveryWriteInItAsync() + { + var transport = new RecordingTransport { CanCoalesceChunks = true, FailIndex = 3 }; + using var channel = new WriteProbeChannel(m_telemetry, transport); + + channel.Write(0, useCollection: true); + await transport.FirstSendStarted.Task.ConfigureAwait(false); + for (int ii = 1; ii < 6; ii++) + { + channel.Write(ii, useCollection: ii % 2 == 0); + } + transport.ReleaseFirstSend.SetResult(true); + await channel.WaitForCompletionsAsync(6).ConfigureAwait(false); + + channel.Write(6, useCollection: false); + await channel.WaitForCompletionsAsync(7).ConfigureAwait(false); + + Assert.That(transport.Sent, Is.EqualTo(s_sentAroundTheFailedBatch)); + Assert.That(channel.Failed, Is.EquivalentTo(s_failedBatch)); + Assert.That(channel.Completed, Is.EqualTo(s_sentAroundTheFailedBatch)); + Assert.That(channel.HasPendingWrites, Is.False); + } + + private static readonly int[] s_sentAroundTheFailedBatch = [0, 6]; + private static readonly int[] s_failedBatch = [1, 2, 3, 4, 5]; private static readonly int[] s_sentAroundTheFailure = [0, 1, 2, 4, 5]; private static readonly int[] s_failed = [3]; @@ -196,6 +282,8 @@ public WriteProbeChannel(ITelemetryContext telemetry, IUaSCByteTransport transpo public ConcurrentQueue Failed { get; } = new(); + public int BytesReported => Volatile.Read(ref m_bytesReported); + public void Write(int index, bool useCollection) { byte[] buffer = BufferManager.TakeBuffer(sizeof(int), "WriteProbeChannel"); @@ -227,18 +315,25 @@ protected override void HandleWriteComplete( ServiceResult result) { base.HandleWriteComplete(buffers, state, bytesWritten, result); + Interlocked.Add(ref m_bytesReported, bytesWritten); (ServiceResult.IsGood(result) ? Completed : Failed).Enqueue((int)state!); } + + private int m_bytesReported; } /// /// Records the index carried by each sent chunk and how many sends overlap. /// The first send can be held until released. /// - private sealed class RecordingTransport : IUaSCByteTransport + private sealed class RecordingTransport : IUaSCByteTransport, IUaSCChunkCoalescing { public bool BlockFirstSend { get; set; } = true; + public bool CanCoalesceChunks { get; set; } + + public int SendCalls => Volatile.Read(ref m_sendCalls); + public int FailIndex { get; set; } = -1; public TaskCompletionSource FirstSendStarted { get; } = @@ -277,13 +372,19 @@ public ValueTask ConnectAsync(Uri url, CancellationToken ct) public ValueTask SendChunkAsync(ReadOnlyMemory chunk, CancellationToken ct) { - return SendAsync(BitConverter.ToInt32(chunk.ToArray(), 0)); + return SendAsync([BitConverter.ToInt32(chunk.ToArray(), 0)]); } public ValueTask SendChunkAsync(BufferCollection buffers, CancellationToken ct) { - ArraySegment segment = buffers[0]; - return SendAsync(BitConverter.ToInt32(segment.Array!, segment.Offset)); + // Each write is one 4-byte segment; a coalesced send carries several. + int[] indexes = new int[buffers.Count]; + for (int ii = 0; ii < indexes.Length; ii++) + { + ArraySegment segment = buffers[ii]; + indexes[ii] = BitConverter.ToInt32(segment.Array!, segment.Offset); + } + return SendAsync(indexes); } public ValueTask> ReceiveChunkAsync(CancellationToken ct) @@ -296,8 +397,10 @@ public void Close() { } - private async ValueTask SendAsync(int index) + private async ValueTask SendAsync(int[] indexes) { + int index = indexes[0]; + Interlocked.Increment(ref m_sendCalls); int concurrent = Interlocked.Increment(ref m_concurrent); int max; while (concurrent > (max = Volatile.Read(ref m_maxConcurrent)) && @@ -324,14 +427,14 @@ private async ValueTask SendAsync(int index) } } - if (index == FailIndex) + if (Array.IndexOf(indexes, FailIndex) >= 0) { throw ServiceResultException.Create(StatusCodes.BadConnectionClosed, "send failed"); } lock (m_sentLock) { - m_sent.Add(index); + m_sent.AddRange(indexes); } } finally @@ -345,6 +448,7 @@ private async ValueTask SendAsync(int index) private int m_concurrent; private int m_maxConcurrent; private int m_started; + private int m_sendCalls; } } } From 1256c5940893952fe3099a3a59cf7caf56e93b25 Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Sun, 11 Oct 2026 21:15:20 +0200 Subject: [PATCH 5/6] TcpByteTransport: fail the stream after an interrupted large-chunk read Review follow-ups: - A receive that fails while reading the body of a chunk larger than the staging buffer has consumed part of the chunk, which goes back with the rented buffer. The transport now remembers that and fails later receives with BadTcpInternalError instead of reading body bytes as the next chunk header. Cancellation while the bytes are in the staging buffer still resumes. - On .NET Framework the staging receive takes no token; the bytes it returned are now kept before the cancellation is observed. The partial-header cancellation tests run where ReceiveAsync is cancellable (.NET 5+). - Tests send their whole payload (SendAsync may send less), and the harness wraps the connect-parallel line. Co-Authored-By: Claude Opus 5.5 --- perf/ServerLoadHarness/Program.cs | 5 +- src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs | 44 +++++++++++--- .../Stack/Transport/TcpByteTransportTests.cs | 57 ++++++++++++++++--- 3 files changed, 88 insertions(+), 18 deletions(-) diff --git a/perf/ServerLoadHarness/Program.cs b/perf/ServerLoadHarness/Program.cs index 83879fc8de..e495e23747 100644 --- a/perf/ServerLoadHarness/Program.cs +++ b/perf/ServerLoadHarness/Program.cs @@ -311,10 +311,13 @@ await client await using ConfiguredAsyncDisposable disposeChannelManager = channelManager.ConfigureAwait(false); var sessions = new ISession[sessionCount]; var connectTime = Stopwatch.StartNew(); + // Shared sessions connect one at a time by default: concurrent creation + // on one ClientChannelManager fails with BadSecureChannelClosed. + int connectParallel = clientKind == "shared" ? ArgInt("connect-parallel", 1) : 32; await Parallel.ForAsync( 0, sessionCount, - new ParallelOptions { MaxDegreeOfParallelism = clientKind == "shared" ? ArgInt("connect-parallel", 1) : 32 }, + new ParallelOptions { MaxDegreeOfParallelism = connectParallel }, async (i, ct) => sessions[i] = clientKind switch { "managed" => await new ManagedSessionBuilder(client.Config, telemetry) diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs index fe8709613f..873abf9c4a 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs @@ -340,6 +340,13 @@ public async ValueTask> ReceiveChunkAsync(CancellationToken c try { + if (m_receiveBroken) + { + throw ServiceResultException.Create( + StatusCodes.BadTcpInternalError, + "An earlier receive was interrupted inside a chunk; the stream cannot be read further."); + } + // An idle peer is waited for with the small staging buffer only: the // receive buffer (and its share of a process-wide memory budget) is // rented once the chunk header has arrived and been validated. One @@ -406,13 +413,24 @@ public async ValueTask> ReceiveChunkAsync(CancellationToken c if (staged < messageSize) { - await ReadExactAsync( - socket, - buffer, - staged, - messageSize - staged, - ct) - .ConfigureAwait(false); + try + { + await ReadExactAsync( + socket, + buffer, + staged, + messageSize - staged, + ct) + .ConfigureAwait(false); + } + catch + { + // Part of the chunk was consumed from the stream and + // goes back with the buffer: the next bytes are not a + // chunk header, so the stream cannot be read again. + m_receiveBroken = true; + throw; + } } } finally @@ -468,7 +486,6 @@ private async ValueTask FillStagingAsync(Socket socket, byte[] staging, Cancella new ArraySegment(staging, m_stagedCount, staging.Length - m_stagedCount), SocketFlags.None) .ConfigureAwait(false); - ct.ThrowIfCancellationRequested(); #endif if (n <= 0) { @@ -477,6 +494,11 @@ private async ValueTask FillStagingAsync(Socket socket, byte[] staging, Cancella "Remote side closed the connection."); } m_stagedCount += n; +#if !NET5_0_OR_GREATER + // The receive cannot be cancelled here; the bytes it returned are kept + // for the next call before the cancellation is observed. + ct.ThrowIfCancellationRequested(); +#endif } private static byte[] AllocateStaging() @@ -714,6 +736,12 @@ private void ShutdownAndDispose(Socket socket) private int m_stagedOffset; private int m_stagedCount; private int m_receiving; + + /// + /// Set when a receive failed after consuming part of a chunk that did not + /// fit the staging buffer; the stream position is lost. + /// + private bool m_receiveBroken; private readonly ILogger m_logger; [System.Diagnostics.CodeAnalysis.SuppressMessage( diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs index 2699ee2d50..8c480339b2 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs @@ -576,9 +576,7 @@ public async Task ReceiveChunkAsyncReturnsChunksSentTogetherOneAtATimeAsync() BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 100) ]; byte[] all = [.. chunks.SelectMany(c => c)]; - await serverSocket - .SendAsync(new ArraySegment(all), SocketFlags.None) - .ConfigureAwait(false); + await SendAllAsync(serverSocket, all, 0, all.Length).ConfigureAwait(false); foreach (byte[] expected in chunks) { @@ -604,9 +602,7 @@ public async Task ReceiveChunkAsyncCompletesAChunkSplitAcrossSendsAsync() Task> receive = client.ReceiveChunkAsync(CancellationToken.None).AsTask(); foreach ((int offset, int count) in new[] { (0, 3), (3, 64), (67, 29) }) { - await serverSocket - .SendAsync(new ArraySegment(all, offset, count), SocketFlags.None) - .ConfigureAwait(false); + await SendAllAsync(serverSocket, all, offset, count).ConfigureAwait(false); await Task.Delay(50).ConfigureAwait(false); } @@ -615,6 +611,9 @@ await serverSocket } } +#if NET5_0_OR_GREATER + // The pending receive can only be cancelled where Socket.ReceiveAsync takes + // a token; on .NET Framework it waits for data or the close of the transport. [Test] public async Task ReceiveChunkAsyncHonoursCancellationAndKeepsStagedBytesAsync() { @@ -635,13 +634,53 @@ await serverSocket Throws.InstanceOf()); // The next receive continues with the bytes that already arrived. - await serverSocket - .SendAsync(new ArraySegment(chunk, 4, chunk.Length - 4), SocketFlags.None) - .ConfigureAwait(false); + await SendAllAsync(serverSocket, chunk, 4, chunk.Length - 4).ConfigureAwait(false); await AssertReceivesAsync(client, chunk).ConfigureAwait(false); } } + [Test] + public async Task CancellingTheBodyReadOfALargeChunkFailsLaterReceivesAsync() + { + const int largeBufferSize = 65536; + m_bufferManager = new BufferManager("TcpByteTransportTests", largeBufferSize, m_telemetry); + (TcpByteTransport client, Socket serverSocket, TcpListener listener) = + await CreateConnectedPairAsync(largeBufferSize).ConfigureAwait(false); + using var _l = new ListenerScope(listener); + using Socket _s = serverSocket; + using (client) + { + // More than the staging buffer arrives, then the rest of the chunk + // does not: the bytes consumed so far cannot be put back. + byte[] chunk = BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 20000); + await SendAllAsync(serverSocket, chunk, 0, 12000).ConfigureAwait(false); + using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300)); + Assert.That( + async () => await client.ReceiveChunkAsync(cts.Token).ConfigureAwait(false), + Throws.InstanceOf()); + + await SendAllAsync(serverSocket, chunk, 12000, chunk.Length - 12000).ConfigureAwait(false); + ServiceResultException ex = Assert.ThrowsAsync( + async () => await client.ReceiveChunkAsync(CancellationToken.None) + .ConfigureAwait(false))!; + Assert.That(ex.StatusCode, Is.EqualTo((uint)StatusCodes.BadTcpInternalError)); + } + } +#endif + + private static async Task SendAllAsync(Socket socket, byte[] data, int offset, int count) + { + while (count > 0) + { + int sent = await socket + .SendAsync(new ArraySegment(data, offset, count), SocketFlags.None) + .ConfigureAwait(false); + Assert.That(sent, Is.GreaterThan(0)); + offset += sent; + count -= sent; + } + } + private async Task AssertReceivesAsync(TcpByteTransport client, byte[] expected) { ArraySegment chunk = await client From fd00ba38b55e440480d5ca0f560836d4b118087c Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Sun, 11 Oct 2026 21:32:53 +0200 Subject: [PATCH 6/6] Tests: cover the ManagedSession lock paths, coalesced send errors and the receive guard Patch coverage follow-up: CreateSession/ActivateSession/CloseSession passthrough cases; a service call waiting while a writer holds the service lock; a call that fails before the inner call starts faulting the returned task and releasing the reader lock; an unexpected exception in a coalesced send reported as BadTcpInternalError for every write in it; a second concurrent ReceiveChunkAsync rejected. Co-Authored-By: Claude Opus 5.5 --- .../Session/ManagedSessionComplianceTests.cs | 83 +++++++++++++++++++ .../Stack/Transport/TcpByteTransportTests.cs | 22 +++++ .../UaSCBinaryChannelWriteQueueTests.cs | 44 ++++++++++ 3 files changed, 149 insertions(+) diff --git a/tests/Opc.Ua.Client.Tests/Session/ManagedSessionComplianceTests.cs b/tests/Opc.Ua.Client.Tests/Session/ManagedSessionComplianceTests.cs index fa2e4f0372..8418963c74 100644 --- a/tests/Opc.Ua.Client.Tests/Session/ManagedSessionComplianceTests.cs +++ b/tests/Opc.Ua.Client.Tests/Session/ManagedSessionComplianceTests.cs @@ -575,6 +575,20 @@ public static IEnumerable ServicePassthroughCases() yield return Case( "Cancel", m => m.CancelAsync(null, 0, default).AsTask()); + yield return Case( + "CreateSession", + m => m.CreateSessionAsync(null, new ApplicationDescription(), + null, null, null, default, default, 60000, 0, + default).AsTask()); + yield return Case( + "ActivateSession", + m => m.ActivateSessionAsync(null, new SignatureData(), + new ArrayOf(), + new ArrayOf(), default, null, + default).AsTask()); + yield return Case( + "CloseSession", + m => m.CloseSessionAsync(null, false, default).AsTask()); } private static TestCaseData Case( @@ -675,6 +689,75 @@ public async Task ReadAsyncBubblesUpExceptionFromInnerSession() Assert.That(response, Is.Not.Null); } + /// + /// While a writer (reconnect, failover) holds the service lock, a + /// service call waits for it instead of taking the reader lock + /// synchronously. + /// + [Test] + public async Task ServiceCallWaitsWhileAWriterHoldsTheServiceLockAsync() + { + m_innerSession.Channel + .Setup(c => c.SendRequestAsync( + It.IsAny(), + It.IsAny())) + .Returns(new ValueTask(new ReadResponse())); + var serviceLock = (AsyncReaderWriterLock)typeof(ManagedSessionClass) + .GetField("m_serviceLock", BindingFlags.NonPublic | BindingFlags.Instance)! + .GetValue(m_managedSession)!; + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + + Task read; + using (await serviceLock.WriterLockAsync(timeout.Token).ConfigureAwait(false)) + { + read = m_managedSession.ReadAsync( + null, 0, TimestampsToReturn.Both, + new ArrayOf(), timeout.Token).AsTask(); + await Task.Delay(100, timeout.Token).ConfigureAwait(false); + Assert.That(read.IsCompleted, Is.False); + } + + Assert.That(await read.WaitAsync(timeout.Token).ConfigureAwait(false), Is.Not.Null); + } + + /// + /// A service call that fails before the inner call starts (no inner + /// session) reports the error through the returned task and releases + /// the reader lock. + /// + [Test] + public async Task ServiceCallWithoutInnerSessionFaultsTheTaskAsync() + { + FieldInfo sessionField = typeof(ManagedSessionClass) + .GetField("m_session", BindingFlags.NonPublic | BindingFlags.Instance)!; + object? inner = sessionField.GetValue(m_managedSession); + ValueTask read; + sessionField.SetValue(m_managedSession, null); + try + { + read = m_managedSession.ReadAsync( + null, 0, TimestampsToReturn.Both, + new ArrayOf(), default); + } + finally + { + sessionField.SetValue(m_managedSession, inner); + } + + Assert.That( + async () => await read.ConfigureAwait(false), + Throws.TypeOf() + .With.Property("StatusCode") + .EqualTo(StatusCodes.BadNotConnected)); + + var serviceLock = (AsyncReaderWriterLock)typeof(ManagedSessionClass) + .GetField("m_serviceLock", BindingFlags.NonPublic | BindingFlags.Instance)! + .GetValue(m_managedSession)!; + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + using AsyncReaderWriterLock.Releaser writer = + await serviceLock.WriterLockAsync(timeout.Token).ConfigureAwait(false); + } + /// /// Connects a managed session through a mock factory so its worker and service gate follow the real lifecycle. /// diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs index 8c480339b2..9ef9f099c1 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs @@ -668,6 +668,28 @@ public async Task CancellingTheBodyReadOfALargeChunkFailsLaterReceivesAsync() } #endif + [Test] + public async Task ASecondConcurrentReceiveIsRejectedAsync() + { + (TcpByteTransport client, Socket serverSocket, TcpListener listener) = + await CreateConnectedPairAsync().ConfigureAwait(false); + using var _l = new ListenerScope(listener); + using Socket _s = serverSocket; + using (client) + { + // The staged bytes belong to one reader; a second receive on the + // same stream fails instead of interleaving with the first. + Task> first = client.ReceiveChunkAsync(CancellationToken.None).AsTask(); + Assert.That( + async () => await client.ReceiveChunkAsync(CancellationToken.None).ConfigureAwait(false), + Throws.InvalidOperationException); + + byte[] chunk = BuildValidChunk(TcpMessageType.Message | TcpMessageType.Final, 24); + await SendAllAsync(serverSocket, chunk, 0, chunk.Length).ConfigureAwait(false); + AssertChunk(await first.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false), chunk); + } + } + private static async Task SendAllAsync(Socket socket, byte[] data, int offset, int count) { while (count > 0) diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs index 7765e940f8..bb4c014ea7 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelWriteQueueTests.cs @@ -233,6 +233,38 @@ public async Task AFailedCoalescedSendIsReportedForEveryWriteInItAsync() Assert.That(channel.HasPendingWrites, Is.False); } + /// + /// An unexpected exception from a coalesced send is reported as an internal + /// error for every write in it, chunks and buffer collections alike. + /// + [Test] + [CancelAfter(30000)] + public async Task AnUnexpectedErrorInACoalescedSendIsReportedForEveryWriteAsync() + { + var transport = new RecordingTransport + { + CanCoalesceChunks = true, + FailIndex = 2, + FailUnexpectedly = true + }; + using var channel = new WriteProbeChannel(m_telemetry, transport); + + channel.Write(0, useCollection: false); + await transport.FirstSendStarted.Task.ConfigureAwait(false); + for (int ii = 1; ii < 5; ii++) + { + channel.Write(ii, useCollection: ii % 2 == 0); + } + transport.ReleaseFirstSend.SetResult(true); + await channel.WaitForCompletionsAsync(5).ConfigureAwait(false); + + Assert.That(channel.Failed, Is.EquivalentTo(s_failedUnexpectedly)); + Assert.That(channel.FailedCodes, Is.All.EqualTo((StatusCode)StatusCodes.BadTcpInternalError)); + Assert.That(channel.BytesReported, Is.EqualTo(sizeof(int))); + Assert.That(channel.HasPendingWrites, Is.False); + } + + private static readonly int[] s_failedUnexpectedly = [1, 2, 3, 4]; private static readonly int[] s_sentAroundTheFailedBatch = [0, 6]; private static readonly int[] s_failedBatch = [1, 2, 3, 4, 5]; private static readonly int[] s_sentAroundTheFailure = [0, 1, 2, 4, 5]; @@ -284,6 +316,8 @@ public WriteProbeChannel(ITelemetryContext telemetry, IUaSCByteTransport transpo public int BytesReported => Volatile.Read(ref m_bytesReported); + public ConcurrentQueue FailedCodes { get; } = new(); + public void Write(int index, bool useCollection) { byte[] buffer = BufferManager.TakeBuffer(sizeof(int), "WriteProbeChannel"); @@ -316,6 +350,10 @@ protected override void HandleWriteComplete( { base.HandleWriteComplete(buffers, state, bytesWritten, result); Interlocked.Add(ref m_bytesReported, bytesWritten); + if (ServiceResult.IsBad(result)) + { + FailedCodes.Enqueue(result.StatusCode); + } (ServiceResult.IsGood(result) ? Completed : Failed).Enqueue((int)state!); } @@ -332,6 +370,8 @@ private sealed class RecordingTransport : IUaSCByteTransport, IUaSCChunkCoalesci public bool CanCoalesceChunks { get; set; } + public bool FailUnexpectedly { get; set; } + public int SendCalls => Volatile.Read(ref m_sendCalls); public int FailIndex { get; set; } = -1; @@ -429,6 +469,10 @@ private async ValueTask SendAsync(int[] indexes) if (Array.IndexOf(indexes, FailIndex) >= 0) { + if (FailUnexpectedly) + { + throw new InvalidOperationException("send failed"); + } throw ServiceResultException.Create(StatusCodes.BadConnectionClosed, "send failed"); }