diff --git a/src/Opc.Ua.Bindings.Https/Tcp/KestrelTcpTransportListener.cs b/src/Opc.Ua.Bindings.Https/Tcp/KestrelTcpTransportListener.cs index 6858dc3e68..8c5d53539d 100644 --- a/src/Opc.Ua.Bindings.Https/Tcp/KestrelTcpTransportListener.cs +++ b/src/Opc.Ua.Bindings.Https/Tcp/KestrelTcpTransportListener.cs @@ -916,7 +916,7 @@ namespace Opc.Ua.Bindings } // Keeps a sessionless channel classified as in use while the // request is processed so admission does not reclaim it. - using IDisposable usage = channel.TrackPendingRequest(); + using TcpListenerChannel.PendingRequestScope usage = channel.TrackPendingRequest(); try { var context = new SecureChannelContext( diff --git a/src/Opc.Ua.Client/Session/Session.cs b/src/Opc.Ua.Client/Session/Session.cs index 2dd5e44d24..702ede8b3f 100644 --- a/src/Opc.Ua.Client/Session/Session.cs +++ b/src/Opc.Ua.Client/Session/Session.cs @@ -602,7 +602,14 @@ namespace Opc.Ua.Client m_serverCertificate = null; m_keepAliveCancellation?.Dispose(); m_keepAliveCancellation = null; - m_engine?.Dispose(); + if (m_engine is IAsyncDisposable asyncEngine) + { + await asyncEngine.DisposeAsync().ConfigureAwait(false); + } + else + { + m_engine?.Dispose(); + } } base.Dispose(disposing); @@ -4118,10 +4125,12 @@ namespace Opc.Ua.Client } } - // clean up. + // clean up. Asynchronously: the synchronous Dispose blocks a thread + // until the subscription engine has stopped, and many sessions closed + // at once that way starve the thread pool their own close needs. if (closeChannel) { - Dispose(); + await DisposeAsync().ConfigureAwait(false); } return result; diff --git a/src/Opc.Ua.Client/Session/Subscription/DefaultSubscriptionEngine.cs b/src/Opc.Ua.Client/Session/Subscription/DefaultSubscriptionEngine.cs index af5b764a68..e94fdcf325 100644 --- a/src/Opc.Ua.Client/Session/Subscription/DefaultSubscriptionEngine.cs +++ b/src/Opc.Ua.Client/Session/Subscription/DefaultSubscriptionEngine.cs @@ -45,7 +45,7 @@ namespace Opc.Ua.Client /// publish pipeline as a drop-in replacement for the /// classic fire-and-forget engine. /// - public sealed class DefaultSubscriptionEngine : ISubscriptionEngine + public sealed class DefaultSubscriptionEngine : ISubscriptionEngine, IAsyncDisposable { /// /// Initializes a new instance of the @@ -187,6 +187,16 @@ namespace Opc.Ua.Client } } + /// + public async ValueTask DisposeAsync() + { + if (!m_disposed) + { + await m_manager.DisposeAsync().ConfigureAwait(false); + m_disposed = true; + } + } + private void ThrowIfDisposed() { #if NET8_0_OR_GREATER diff --git a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs index 1aa9a6d330..7cffd3e959 100644 --- a/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs +++ b/src/Opc.Ua.Core/Stack/Server/FairRequestQueue.cs @@ -1001,17 +1001,16 @@ namespace Opc.Ua /// public void Dispose() { - CancellationTokenRegistration registration; - lock (m_registrationGate) + var prior = (RegistrationState)Interlocked.Exchange( + ref m_registrationState, (int)RegistrationState.Disposed); + if (prior == RegistrationState.Disposed) { - if (m_disposed) - { - return; - } - m_disposed = true; - registration = m_registration; + return; + } + if (prior == RegistrationState.Registered) + { + m_registration.Dispose(); } - registration.Dispose(); ReleaseQueueLease(); ReleaseExecution(); Interlocked.Exchange(ref m_parkedLease, null)?.Dispose(); @@ -1040,16 +1039,14 @@ namespace Opc.Ua internal void RegisterCancellation() { CancellationTokenRegistration registration = CancellationToken.Register(() => m_queue.Cancel(this)); - bool disposed; - lock (m_registrationGate) - { - disposed = m_disposed; - if (!disposed) - { - m_registration = registration; - } - } - if (disposed) + m_registration = registration; + + // published unless the entry was disposed meanwhile, possibly by the + // callback running inside Register; the registration is ours to end then. + if (Interlocked.CompareExchange( + ref m_registrationState, + (int)RegistrationState.Registered, + (int)RegistrationState.None) != (int)RegistrationState.None) { registration.Dispose(); } @@ -1060,11 +1057,6 @@ namespace Opc.Ua /// private readonly FairRequestQueue m_queue; - /// - /// Serializes registration transfer against entry disposal. - /// - private readonly Lock m_registrationGate = new(); - /// /// The owning FIFO while linked. /// @@ -1091,14 +1083,37 @@ namespace Opc.Ua private IDisposable? m_parkedLease; /// - /// Registration owned until disposal starts. + /// Registration owned until disposal starts; published by + /// . /// private CancellationTokenRegistration m_registration; /// - /// Whether registration ownership has been released. + /// A , stored as an so that + /// can change it; an entry needs no lock object. /// - private bool m_disposed; + private int m_registrationState; + + /// + /// Who owns the cancellation registration of an entry. + /// + private enum RegistrationState + { + /// + /// Not registered yet. + /// + None, + + /// + /// Registered; disposal ends the registration. + /// + Registered, + + /// + /// Disposal started; a registration made afterwards is ended by its maker. + /// + Disposed + } } /// diff --git a/src/Opc.Ua.Core/Stack/Tcp/ChannelGate.cs b/src/Opc.Ua.Core/Stack/Tcp/ChannelGate.cs index c6e686b57a..12bdb4afa7 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/ChannelGate.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/ChannelGate.cs @@ -64,32 +64,26 @@ namespace Opc.Ua.Bindings /// /// It is deliberately not disposable. It replaces a monitor, which has /// no disposed state, and channel teardown enters it and then continues to run - /// paths that enter it again. Disposing the underlying semaphore would turn - /// those into where they previously - /// worked. Nothing leaks: the semaphore only allocates an operating system - /// wait handle if is read, and - /// it never is. + /// paths that enter it again. It holds no operating system resource. /// /// - /// What it costs. Measured by ChannelGateBenchmarks, an - /// uncontended acquisition costs what the underlying - /// costs and allocates nothing, against about - /// 13 ns for the monitor it replaced. The per-message work it is taken - /// around — SymmetricChannelCryptoBenchmarks.EncryptSignThenDecryptVerify - /// — is about 10,500 ns, so the gate does not register against it. The - /// asynchronous entry completes synchronously when uncontended, which a test - /// asserts, so it costs no suspension and no task. + /// What it costs. An uncontended entry and exit are one interlocked + /// compare-and-exchange each and allocate nothing. Only a contended entry + /// takes a lock, to queue a waiter, and only an exit with waiters queued takes + /// it to hand the gate to the first of them. A , + /// which this used to wrap, takes its monitor on every wait and release; the + /// receive loop takes the gate for every message, and once that monitor had + /// seen contention every entry went through its slow path. The asynchronous + /// entry completes synchronously when uncontended, which a test asserts, so + /// it costs no suspension and no task. + /// + /// + /// Waiters are served first in, first out, and an exit with waiters hands the + /// gate straight to the first one, so a new entrant cannot overtake them. A + /// waiter resumes on the thread pool, never on the thread that left the gate. /// /// - // CA1001: the semaphore is deliberately not disposed. This type replaces a - // monitor, which has no disposed state, and channel teardown enters the gate - // and then runs paths that enter it again; disposing would turn those into - // ObjectDisposedException where they previously worked. SemaphoreSlim only - // allocates an operating system wait handle when AvailableWaitHandle is read, - // which this type never does, so there is nothing to release. -#pragma warning disable CA1001 internal sealed class ChannelGate -#pragma warning restore CA1001 { /// /// Enters the gate, blocking until it is free. @@ -102,7 +96,12 @@ namespace Opc.Ua.Bindings /// public Releaser Enter() { - m_semaphore.Wait(); + if (Interlocked.CompareExchange(ref m_state, kHeld, kFree) != kFree) + { + Waiter? waiter = QueueWaiter(); + // a queued waiter is completed only by the exit that hands it the gate. + waiter?.Task.GetAwaiter().GetResult(); + } return new Releaser(this); } @@ -114,18 +113,25 @@ namespace Opc.Ua.Bindings /// A handle that leaves the gate when disposed. public ValueTask EnterAsync(CancellationToken ct = default) { - Task wait = m_semaphore.WaitAsync(ct); + if (ct.IsCancellationRequested) + { + return new ValueTask(Task.FromCanceled(ct)); + } // Uncontended is the common case. Completing synchronously keeps the // caller's sequencing identical to the lock this replaces. - // Task.IsCompletedSuccessfully does not exist on the .NET Framework - // targets, so the status is read directly. - if (wait.Status == TaskStatus.RanToCompletion) + if (Interlocked.CompareExchange(ref m_state, kHeld, kFree) == kFree) { return new ValueTask(new Releaser(this)); } - return AwaitEntryAsync(wait); + Waiter? waiter = QueueWaiter(); + if (waiter == null) + { + return new ValueTask(new Releaser(this)); + } + + return AwaitEntryAsync(waiter, ct); } /// @@ -134,21 +140,234 @@ namespace Opc.Ua.Bindings /// /// Used by tests that need to establish contention before entering. /// - internal bool IsHeldBySomeContextForTest => m_semaphore.CurrentCount == 0; + internal bool IsHeldBySomeContextForTest => Volatile.Read(ref m_state) != kFree; - private async ValueTask AwaitEntryAsync(Task wait) + private async ValueTask AwaitEntryAsync(Waiter waiter, CancellationToken ct) { - await wait.ConfigureAwait(false); + CancellationTokenRegistration registration = default; + if (ct.CanBeCanceled) + { + // runs at once when the token is already cancelled. + registration = ct.Register(static state => ((Waiter)state!).Cancel(), waiter); + } + + try + { + await waiter.Task.ConfigureAwait(false); + } + finally + { + registration.Dispose(); + } return new Releaser(this); } - private void Leave() + /// + /// Queues a waiter for a gate that is held, or takes the gate when it was + /// left in the meantime. + /// + /// The waiter, or when the caller took the + /// gate. + private Waiter? QueueWaiter() { - m_semaphore.Release(); + lock (m_waitersLock) + { + while (true) + { + int state = Volatile.Read(ref m_state); + if (state == kFree) + { + if (Interlocked.CompareExchange(ref m_state, kHeld, kFree) == kFree) + { + return null; + } + } + else if (state == kContended || + Interlocked.CompareExchange(ref m_state, kContended, kHeld) == kHeld) + { + // from here on the exit cannot leave the gate without the lock. + break; + } + } + + var waiter = new Waiter(this); + Append(waiter); + return waiter; + } } - private readonly SemaphoreSlim m_semaphore = new(1, 1); + /// + /// Leaves the gate, or hands it to the first waiter. + /// + private void Leave() + { + int state = Interlocked.CompareExchange(ref m_state, kFree, kHeld); + if (state == kHeld) + { + return; + } + + if (state == kFree) + { + throw new SynchronizationLockException("The channel gate is not held."); + } + + Waiter? next; + lock (m_waitersLock) + { + // waiters are queued while the state is contended; a cancelled one + // is unlinked, and the last one to go returns the state to held, + // possibly after the exchange above saw it contended. + next = m_first; + if (next == null) + { + Volatile.Write(ref m_state, kFree); + return; + } + Unlink(next); + if (m_first == null) + { + Volatile.Write(ref m_state, kHeld); + } + } + + // the gate stays held: ownership passes to the waiter. + next.TrySetResult(true); + } + + /// + /// Removes a waiter whose wait was cancelled, unless it has already been + /// handed the gate. + /// + /// when the waiter was removed. + private bool Remove(Waiter waiter) + { + lock (m_waitersLock) + { + if (!waiter.Queued) + { + return false; + } + + Unlink(waiter); + if (m_first == null) + { + // the holder still holds the gate, now without waiters. + Volatile.Write(ref m_state, kHeld); + } + return true; + } + } + + /// + /// Appends a waiter; called under . + /// + private void Append(Waiter waiter) + { + waiter.Previous = m_last; + if (m_last == null) + { + m_first = waiter; + } + else + { + m_last.Next = waiter; + } + m_last = waiter; + waiter.Queued = true; + } + + /// + /// Unlinks a queued waiter; called under . + /// + private void Unlink(Waiter waiter) + { + if (waiter.Previous == null) + { + m_first = waiter.Next; + } + else + { + waiter.Previous.Next = waiter.Next; + } + + if (waiter.Next == null) + { + m_last = waiter.Previous; + } + else + { + waiter.Next.Previous = waiter.Previous; + } + + waiter.Next = null; + waiter.Previous = null; + waiter.Queued = false; + } + + /// + /// The gate is free. + /// + private const int kFree = 0; + + /// + /// The gate is held and no waiter is queued; it is left without the lock. + /// + private const int kHeld = 1; + + /// + /// The gate is held and waiters are queued; leaving it takes the lock. + /// + private const int kContended = 2; + + /// + /// , or . + /// The change to and from is made under + /// , the others with a compare-and-exchange. + /// + private int m_state; + + /// + /// Protects the waiter queue. + /// + private readonly Lock m_waitersLock = new(); + + private Waiter? m_first; + private Waiter? m_last; + + /// + /// A queued entry, completed with the gate or cancelled. + /// + private sealed class Waiter : TaskCompletionSource + { + public Waiter(ChannelGate gate) + : base(TaskCreationOptions.RunContinuationsAsynchronously) + { + m_gate = gate; + } + + /// + /// Whether the waiter is in the queue; changed under the gate's lock. + /// + public bool Queued; + + public Waiter? Next; + public Waiter? Previous; + + /// + /// Cancels the wait, unless the gate was handed over already. + /// + public void Cancel() + { + if (m_gate.Remove(this)) + { + TrySetCanceled(); + } + } + + private readonly ChannelGate m_gate; + } /// /// Leaves the gate when disposed. diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs index 873abf9c4a..87ed838061 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs @@ -98,7 +98,6 @@ namespace Opc.Ua.Bindings m_bufferManager = bufferManager ?? throw new ArgumentNullException(nameof(bufferManager)); m_receiveBufferSize = receiveBufferSize; m_logger = telemetry.CreateLogger(); - m_sendLock = new SemaphoreSlim(1, 1); } /// @@ -232,15 +231,10 @@ namespace Opc.Ua.Bindings using CancellationTokenSource? linkedCts = ct.CanBeCanceled ? CancellationTokenSource.CreateLinkedTokenSource(ct, m_sendCancellation.Token) : null; - await m_sendLock.WaitAsync(linkedCts?.Token ?? m_sendCancellation.Token).ConfigureAwait(false); - try + using (await m_sendLock.EnterAsync(linkedCts?.Token ?? m_sendCancellation.Token).ConfigureAwait(false)) { await SendMemoryAsync(socket, chunk, ct).ConfigureAwait(false); } - finally - { - m_sendLock.Release(); - } } /// @@ -257,8 +251,7 @@ namespace Opc.Ua.Bindings using CancellationTokenSource? linkedCts = ct.CanBeCanceled ? CancellationTokenSource.CreateLinkedTokenSource(ct, m_sendCancellation.Token) : null; - await m_sendLock.WaitAsync(linkedCts?.Token ?? m_sendCancellation.Token).ConfigureAwait(false); - try + using (await m_sendLock.EnterAsync(linkedCts?.Token ?? m_sendCancellation.Token).ConfigureAwait(false)) { if (buffers.Count == 1) { @@ -281,10 +274,6 @@ namespace Opc.Ua.Bindings static (s, pending) => s.SendAsync(pending, SocketFlags.None)) .ConfigureAwait(false); } - finally - { - m_sendLock.Release(); - } } #if NET6_0_OR_GREATER @@ -744,11 +733,11 @@ namespace Opc.Ua.Bindings private bool m_receiveBroken; private readonly ILogger m_logger; - [System.Diagnostics.CodeAnalysis.SuppressMessage( - "Usage", - "CA2213:Disposable fields should be disposed", - Justification = "The semaphore must remain undisposed so queued send waiters can observe transport cancellation and unwind.")] - private readonly SemaphoreSlim m_sendLock; + /// + /// Serialises the sends. The channel has one writer, so it is uncontended + /// there and costs one interlocked exchange each way. + /// + private readonly ChannelGate m_sendLock = new(); [System.Diagnostics.CodeAnalysis.SuppressMessage( "Usage", diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs index c46443951d..ecfbabb022 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpListenerChannel.cs @@ -357,10 +357,10 @@ namespace Opc.Ua.Bindings /// /// Keeps a decoded sessionless request classified as in-use through dispatch and asynchronous processing. /// - internal IDisposable TrackPendingRequest() + internal PendingRequestScope TrackPendingRequest() { Interlocked.Increment(ref m_pendingServiceRequests); - return new PendingRequest(() => Interlocked.Decrement(ref m_pendingServiceRequests)); + return new PendingRequestScope(this); } /// @@ -1042,11 +1042,17 @@ namespace Opc.Ua.Bindings /// /// Sends a fault response secured with the symmetric keys. /// - /// The token that secures the fault. + /// The token of the failed request. The fault is + /// secured with the token that is current when the writer sends it, which + /// is this one or the one the client activated after it. /// The request id of the failed request. /// The fault to report. /// The RequestHandle of the failed request, echoed in /// the ResponseHeader as OPC 10000-4 §7.33 recommends; 0 if it is unknown. + /// + /// Needs no lock: the fault is encoded here and queued for the writer, which + /// numbers and secures it (see ). + /// protected void SendServiceFault( ChannelToken token, uint requestId, @@ -1055,7 +1061,8 @@ namespace Opc.Ua.Bindings { m_logger.TcpListenChannelLog7(ChannelId, requestId, fault.StatusCode); - BufferCollection? buffers = null; + _ = token; + EncodedSymmetricMessage? encoded = null; try { @@ -1079,22 +1086,25 @@ namespace Opc.Ua.Bindings // the limits should never be exceeded when sending a fault. - // secure message. - buffers = WriteSymmetricMessage( + EncodedSymmetricMessage message = EncodeSymmetricMessage( TcpMessageType.Message, requestId, - token, response, false, - out bool limitsExceeded); + out _); + encoded = message; - // send message. - BeginWriteMessage(buffers, null); - buffers = null; + // queue for the writer, which secures it. Without a transport the + // fault is dropped, as a write to a closed transport would be. + if (BeginWriteSymmetricMessage(message, null)) + { + encoded = null; + } } catch (Exception e) { - buffers?.Release(BufferManager, "SendServiceFault"); + encoded?.Chunks.Release(BufferManager, "SendServiceFault"); + encoded = null; m_logger.TcpListenChannelLog8(e, ChannelId, requestId, fault.StatusCode); @@ -1104,6 +1114,10 @@ namespace Opc.Ua.Bindings StatusCodes.BadTcpInternalError, "Unexpected error sending a service fault.")); } + finally + { + encoded?.Chunks.Release(BufferManager, "SendServiceFault"); + } } /// @@ -1242,23 +1256,32 @@ namespace Opc.Ua.Bindings } /// - /// Keeps a decoded request in use until its owner releases the pending-request count once. + /// Keeps a decoded request in use until its owner releases the pending-request count. /// - /// Callback that returns this request's pending count. - private sealed class PendingRequest(Action release) : IDisposable + /// + /// A struct, so tracking a request allocates nothing. Every use is a + /// declaration over the value returned by + /// , which disposes it exactly once. + /// + internal readonly struct PendingRequestScope : IDisposable { - /// - /// Returns the pending-request count at most once. - /// - public void Dispose() + internal PendingRequestScope(TcpListenerChannel channel) { - Interlocked.Exchange(ref m_release, null)?.Invoke(); + m_channel = channel; } /// - /// Pending-count release callback, cleared atomically when consumed. + /// Returns the pending-request count. /// - private Action? m_release = release; + public void Dispose() + { + if (m_channel != null) + { + Interlocked.Decrement(ref m_channel.m_pendingServiceRequests); + } + } + + private readonly TcpListenerChannel? m_channel; } private readonly ILogger m_logger; diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs index f2d0dc81e0..0373b96583 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs @@ -366,32 +366,41 @@ namespace Opc.Ua.Bindings throw new ObjectDisposedException(nameof(TcpServerChannel)); } - // put channel back in open state. + // Retire the old loop BEFORE closing the socket it reads + // from. Closing first makes that loop fail out of + // ReceiveChunkAsync with a socket error rather than a + // cancellation, and its error path faults the channel - on + // the new transport, after this method has already reported + // the reconnect as successful. Detached before the new token + // is activated, nothing still queued for the old transport is + // secured with that token: the writer drops it unnumbered, and + // responses produced meanwhile wait for the gate. + dropped = DetachTransport(); + ActivateToken(token); - State = TcpChannelState.Open; if (transport is IUaSCByteTransportLimits transportLimits) { transportLimits.SetReceiveBufferSize(ReceiveBufferSize); } - // Retire the old loop BEFORE closing the socket it reads - // from. Closing first makes that loop fail out of - // ReceiveChunkAsync with a socket error rather than a - // cancellation, and its error path faults the channel - on - // the new transport, after this method has already reported - // the reconnect as successful. - dropped = DetachTransport(); - - Transport = transport; if (Volatile.Read(ref m_disposed) != 0) { throw new ObjectDisposedException(nameof(TcpServerChannel)); } - StartReceiveLoop(); - // send response. - SendOpenSecureChannelResponse(requestId, token, request, true); + // The transport and the response change together under the + // send order lock, so the writer numbers nothing in between: + // the response is the first message on the new transport. + using (SendOrder.Enter()) + { + Transport = transport; + StartReceiveLoop(); + SendOpenSecureChannelResponse(requestId, token, request, true); + } + + // put channel back in open state. + State = TcpChannelState.Open; // send any queued responses. ResetQueuedResponses(OnChannelReconnected); @@ -427,7 +436,7 @@ namespace Opc.Ua.Bindings ArraySegment messageChunk, CancellationToken ct) { - using IDisposable usage = TrackPendingRequest(); + using PendingRequestScope usage = TrackPendingRequest(); PendingRequestDispatch? pending = null; bool ownsBuffer; @@ -503,11 +512,11 @@ namespace Opc.Ua.Bindings // Outside the gate on purpose: the handler is application code, and // its continuation may call SendResponse straight back on this // stack. See ProcessRequestMessage. - if (pending != null) + if (pending is PendingRequestDispatch dispatch) { try { - RequestReceived?.Invoke(this, pending.RequestId, pending.Request); + RequestReceived?.Invoke(this, dispatch.RequestId, dispatch.Request); } catch (Exception e) { @@ -519,7 +528,7 @@ namespace Opc.Ua.Bindings } finally { - pending.Chunks?.Release(BufferManager, "ProcessRequestMessage"); + dispatch.Chunks?.Release(BufferManager, "ProcessRequestMessage"); } } @@ -1259,19 +1268,24 @@ namespace Opc.Ua.Bindings // key served over a network still occupies a thread; that is recorded // as a residual limitation rather than cascading async through eight // fault call sites. - chunksToSend = WriteAsymmetricMessage( - TcpMessageType.Open, - requestId, - ServerCertificate, - senderCertificateChain: null, - ClientCertificate, - new ArraySegment(buffer, 0, buffer.Length), - !renew ? m_oscRequestSignature : null, - out _); + // Numbered and queued under SendOrder, ahead of the responses the + // writer has not numbered yet. + using (SendOrder.Enter()) + { + chunksToSend = WriteAsymmetricMessage( + TcpMessageType.Open, + requestId, + ServerCertificate, + senderCertificateChain: null, + ClientCertificate, + new ArraySegment(buffer, 0, buffer.Length), + !renew ? m_oscRequestSignature : null, + out _); - // write the message to the server. - BeginWriteMessage(chunksToSend, null); - chunksToSend = null; + // write the message to the server. + BeginWriteMessage(chunksToSend, null); + chunksToSend = null; + } } catch (Exception e) { @@ -1300,8 +1314,9 @@ namespace Opc.Ua.Bindings /// /// Used by the synchronous reconnect handoff, which is reached through a /// public override that cannot become asynchronous without breaking the - /// contract. Signing here with a key served over a network still occupies - /// a thread; the open path proper uses + /// contract, and which holds + /// while it calls this. Signing here with a key served over a network still + /// occupies a thread; the open path proper uses /// and does not. /// private void SendOpenSecureChannelResponse( @@ -1325,6 +1340,8 @@ namespace Opc.Ua.Bindings byte[] buffer = BinaryEncoder.EncodeMessage(response, Quotas.MessageContext); + // numbered and queued under SendOrder, which the caller holds; see + // SendOpenSecureChannelResponseAsync. BufferCollection chunksToSend = WriteAsymmetricMessage( TcpMessageType.Open, requestId, @@ -1374,22 +1391,30 @@ namespace Opc.Ua.Bindings byte[] buffer = BinaryEncoder.EncodeMessage(response, Quotas.MessageContext); - AsymmetricWriteResult written = await WriteAsymmetricMessageAsync( - TcpMessageType.Open, - requestId, - ServerCertificate, - ServerCertificateChain, - ClientCertificate, - new ArraySegment(buffer, 0, buffer.Length), - !renew ? m_oscRequestSignature : null, - ct).ConfigureAwait(false); - - if (!renew) + // A renewal is answered while responses are in flight. The response takes + // its sequence number here, so it is numbered and queued under SendOrder: + // the writer sends it ahead of the responses it has not numbered yet. The + // writer waits while the response is signed, as responses waited for the + // gate before. + using (await SendOrder.EnterAsync(ct).ConfigureAwait(false)) { - ChannelThumbprint = written.Signature; - } + AsymmetricWriteResult written = await WriteAsymmetricMessageAsync( + TcpMessageType.Open, + requestId, + ServerCertificate, + ServerCertificateChain, + ClientCertificate, + new ArraySegment(buffer, 0, buffer.Length), + !renew ? m_oscRequestSignature : null, + ct).ConfigureAwait(false); - SendOpenSecureChannelChunks(written.Chunks); + if (!renew) + { + ChannelThumbprint = written.Signature; + } + + SendOpenSecureChannelChunks(written.Chunks); + } } /// @@ -1515,7 +1540,7 @@ namespace Opc.Ua.Bindings /// Stream decoding copies the request's strings and byte strings; an asynchronous /// callback may continue after these buffers are released. /// - private sealed record PendingRequestDispatch( + private readonly record struct PendingRequestDispatch( uint RequestId, IServiceRequest Request, BufferCollection? Chunks); @@ -1676,10 +1701,7 @@ namespace Opc.Ua.Bindings } // decode the request. - using var serviceRequestStream = new ArraySegmentStream(chunksToProcess); - IServiceRequest request = BinaryDecoder.DecodeMessage( - serviceRequestStream, - Quotas.MessageContext); + IServiceRequest request = DecodeRequest(chunksToProcess); if (request == null) { SendServiceFault( @@ -1742,6 +1764,26 @@ namespace Opc.Ua.Bindings } } + /// + /// Decodes the request in the chunks of a message. + /// + /// + /// A request in one chunk, which most are, is decoded from that chunk's + /// buffer without a stream over the chunk list. Either decoder copies the + /// values it reads, so the request does not borrow the pooled buffers. + /// + private IServiceRequest DecodeRequest(BufferCollection chunks) + { + if (chunks.Count == 1) + { + using var decoder = new BinaryDecoder(chunks[0], Quotas.MessageContext); + return decoder.DecodeMessage(); + } + + using var stream = new ArraySegmentStream(chunks); + return BinaryDecoder.DecodeMessage(stream, Quotas.MessageContext); + } + /// /// Reads the RequestHandle of a request message that could not be /// processed, for the ServiceFault that reports it. @@ -1772,6 +1814,11 @@ namespace Opc.Ua.Bindings throw new ArgumentNullException(nameof(response)); } + if (TrySendResponseWithoutGate(requestId, response)) + { + return false; + } + using (Gate.Enter()) { return SendResponseCore(requestId, response); @@ -1783,11 +1830,11 @@ namespace Opc.Ua.Bindings /// without blocking the calling thread. /// /// - /// Used by the request dispatch path, which runs on thread-pool workers. With many - /// requests in flight on one channel, the synchronous - /// parked one pool thread per waiting response on the gate; the receive loop that - /// was handed the gate then waited for a free pool thread to continue, and the - /// channel stalled until the pool injected more threads. + /// Used by the request dispatch path, which runs on thread-pool workers. An open + /// channel takes no lock for a response: it is encoded on the caller's thread and + /// queued for the writer, which numbers and secures it (see + /// ). Only a channel that is not open, or + /// whose renewed token is due for activation, waits for the gate. /// /// true if the response was retained for delivery after reconnect. /// is null. @@ -1799,12 +1846,92 @@ namespace Opc.Ua.Bindings throw new ArgumentNullException(nameof(response)); } + if (TrySendResponseWithoutGate(requestId, response)) + { + return false; + } + using (await Gate.EnterAsync().ConfigureAwait(false)) { return SendResponseCore(requestId, response); } } + /// + /// Encodes and queues a response without the channel gate, when the channel + /// is open and no token change is due. + /// + /// The request being answered. + /// The response. + /// when the caller must send the response + /// under the gate instead: the channel is not open, a renewed token is due + /// for activation, the transport is gone, or the response cannot be encoded + /// (the gate path sends the fault). Nothing was sent then. + /// + /// The state is read without the gate, so the channel can fault or close + /// right after the check. The response then goes to the transport the + /// channel had, and is lost with it, like a response whose write was queued + /// just before the channel faulted always was. + /// + private bool TrySendResponseWithoutGate( + uint requestId, + IServiceResponse response) + { + if (State != TcpChannelState.Open || + IsRenewedTokenActivationDue || + Transport == null) + { + return false; + } + + EncodedSymmetricMessage encoded; + try + { + // note that the server does nothing if the message limits are exceeded. + encoded = EncodeSymmetricMessage( + TcpMessageType.Message, + requestId, + response, + false, + out _); + } + catch (Exception) + { + // the gate path encodes again and reports the failure. + return false; + } + + if (!BeginWriteSymmetricMessage(encoded, null)) + { + encoded.Chunks.Release(BufferManager, "SendResponse"); + return false; + } + + m_eventLogger.CoreSendResponse((int)ChannelId, (int)requestId); + TrackSessions(response); + return true; + } + + /// + /// Counts the sessions activated and closed on the channel, when no session + /// binding provider does. + /// + private void TrackSessions(IServiceResponse response) + { + if (Quotas.SessionBindingProvider == null && + response is ActivateSessionResponse activateSessionResponse && + StatusCode.IsGood(activateSessionResponse.ResponseHeader.ServiceResult)) + { + AddSession(); + } + else if (Quotas.SessionBindingProvider == null && + response is CloseSessionResponse closeSessionResponse && + StatusCode.IsGood(closeSessionResponse.ResponseHeader.ServiceResult)) + { + RemoveSession(); + } + } + /// /// Encodes and queues a response while the caller holds the channel gate. /// @@ -1827,23 +1954,21 @@ namespace Opc.Ua.Bindings } m_eventLogger.CoreSendResponse((int)ChannelId, (int)requestId); - BufferCollection? buffers = null; // a response sent long after its request (Publish) must not be secured // with a token that has expired while the client sent nothing. ActivateRenewedTokenIfDue(); + EncodedSymmetricMessage encoded; try { // note that the server does nothing if the message limits are exceeded. - - buffers = WriteSymmetricMessage( + encoded = EncodeSymmetricMessage( TcpMessageType.Message, requestId, - CurrentToken!, response, false, - out bool limitsExceeded); + out _); } catch (Exception e) { @@ -1859,32 +1984,16 @@ namespace Opc.Ua.Bindings return false; } - try + // queued for the writer, which secures it with the token current then. + if (!BeginWriteSymmetricMessage(encoded, null)) { - BeginWriteMessage(buffers, null); - buffers = null!; - } - catch (Exception) - { - buffers?.Release(BufferManager, "SendResponse"); + encoded.Chunks.Release(BufferManager, "SendResponse"); m_queuedResponses[requestId] = response; return true; } - if (Quotas.SessionBindingProvider == null && - response is ActivateSessionResponse activateSessionResponse && - StatusCode.IsGood(activateSessionResponse.ResponseHeader.ServiceResult)) - { - AddSession(); - } - else if (Quotas.SessionBindingProvider == null && - response is CloseSessionResponse closeSessionResponse && - StatusCode.IsGood(closeSessionResponse.ResponseHeader.ServiceResult)) - { - RemoveSession(); - } - + TrackSessions(response); return false; } } diff --git a/src/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs b/src/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs index 8bf1306314..7b4f9a67fa 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/TcpTransportListener.cs @@ -1710,7 +1710,7 @@ namespace Opc.Ua.Bindings uint requestId, IServiceRequest request) { - using IDisposable usage = channel.TrackPendingRequest(); + using TcpListenerChannel.PendingRequestScope usage = channel.TrackPendingRequest(); IServiceResponse? response = null; bool responseRetained = false; try diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.Symmetric.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.Symmetric.cs index 5390a69466..593f381959 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.Symmetric.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.Symmetric.cs @@ -32,6 +32,7 @@ using System.Buffers.Binary; using System.IO; using System.Security.Cryptography; using System.Text; +using System.Threading; using Microsoft.Extensions.Logging; namespace Opc.Ua.Bindings @@ -200,18 +201,27 @@ namespace Opc.Ua.Bindings } m_renewedTokenWithKeys = null; - // a pending renewal superseded by another token (reconnect) is dropped. - if (RenewedToken != null && !ReferenceEquals(RenewedToken, token)) + ChannelToken? previousToken; + + // the writer secures messages with the current token under the send + // order lock; swapping it there keeps a token from being disposed while + // a chunk is secured with it. + using (SendOrder.Enter()) { - RenewedToken.Dispose(); + // a pending renewal superseded by another token (reconnect) is dropped. + if (RenewedToken != null && !ReferenceEquals(RenewedToken, token)) + { + RenewedToken.Dispose(); + } + + PreviousToken?.Dispose(); + PreviousToken = CurrentToken; + CurrentToken = token; + RenewedToken = null; + previousToken = PreviousToken; } - PreviousToken?.Dispose(); - PreviousToken = CurrentToken; - CurrentToken = token; - RenewedToken = null; - - TokenActivatedCallback?.Invoke(token, PreviousToken); + TokenActivatedCallback?.Invoke(token, previousToken); if (m_logger.IsEnabled(LogLevel.Information)) { @@ -278,17 +288,34 @@ namespace Opc.Ua.Bindings } } + /// + /// Whether would activate a token. + /// + /// + /// Read without the gate by a sender deciding whether it may skip it. The + /// answer can be stale by the time the message is sealed, which only moves + /// the activation to the next message or the next received chunk. + /// + private protected bool IsRenewedTokenActivationDue => + RenewedToken != null && + CurrentToken is ChannelToken currentToken && + currentToken.IsActivationRequired(TimeProvider); + /// /// Discards the tokens. /// protected void DiscardTokens() { - PreviousToken?.Dispose(); - PreviousToken = null; - CurrentToken?.Dispose(); - CurrentToken = null; - RenewedToken?.Dispose(); - RenewedToken = null; + // see ActivateToken. + using (SendOrder.Enter()) + { + PreviousToken?.Dispose(); + PreviousToken = null; + CurrentToken?.Dispose(); + CurrentToken = null; + RenewedToken?.Dispose(); + RenewedToken = null; + } TokenActivatedCallback?.Invoke(null, null); } @@ -536,6 +563,13 @@ namespace Opc.Ua.Bindings /// /// Secures the message using the security token. /// + /// + /// Encodes and seals in one step. The sequence numbers are taken under + /// , but the caller queues the chunks after it is + /// released, so this is only correct on a channel that queues no unsealed + /// messages () and serialises its + /// numbered writes otherwise, as the client channel does under its gate. + /// protected BufferCollection WriteSymmetricMessage( uint messageType, uint requestId, @@ -543,6 +577,89 @@ namespace Opc.Ua.Bindings object messageBody, bool isRequest, out bool limitsExceeded) + { + EncodedSymmetricMessage encoded = EncodeSymmetricMessage( + messageType, + requestId, + messageBody, + isRequest, + out limitsExceeded); + + using (SendOrder.Enter()) + { + return SealSymmetricMessage(encoded, token); + } + } + + /// + /// A symmetric message encoded into chunks whose headers, sequence numbers + /// and security are not written yet. + /// + /// + /// Encoding needs nothing that is ordered per channel, so any thread can do + /// it without the gate. Sealing takes the token and the sequence numbers and + /// is done by the writer, in queue order (see ). + /// + private protected readonly struct EncodedSymmetricMessage + { + public EncodedSymmetricMessage( + uint messageType, + uint requestId, + bool isRequest, + bool aborted, + BufferCollection chunks) + { + MessageType = messageType; + RequestId = requestId; + IsRequest = isRequest; + Aborted = aborted; + Chunks = chunks; + } + + /// + /// The message type without the chunk type. + /// + public uint MessageType { get; } + + /// + /// The request the message belongs to. + /// + public uint RequestId { get; } + + /// + /// Whether the message is a request (client keys secure it). + /// + public bool IsRequest { get; } + + /// + /// Whether the last chunk is an abort chunk replacing the rest of the + /// message because it exceeded the limits. + /// + public bool Aborted { get; } + + /// + /// The chunks, each holding its body after the reserved header space. + /// Owned by the message until sealed. + /// + public BufferCollection Chunks { get; } + } + + /// + /// Encodes a message into chunks without securing them. + /// + /// The message type without the chunk type. + /// The request the message belongs to. + /// An encodeable, or the raw bytes to send. + /// Whether the message is a request. + /// Set when the message was cut off with an + /// abort chunk because it exceeds the negotiated limits. + /// The encoded message. The caller owns its chunks. + private protected EncodedSymmetricMessage EncodeSymmetricMessage( + uint messageType, + uint requestId, + object messageBody, + bool isRequest, + out bool limitsExceeded) { limitsExceeded = false; bool success = false; @@ -559,11 +676,7 @@ namespace Opc.Ua.Bindings int maxPlainTextSize = maxCipherBlocks * EncryptionBlockSize; int signatureSize = SymmetricSignatureSize; - // token.SecurityPolicy is set in ComputeKeys before the channel sends messages. - int paddingCountSize = - SecurityMode != MessageSecurityMode.SignAndEncrypt || token.SecurityPolicy!.NoSymmetricEncryptionPadding - ? 0 - : (EncryptionBlockSize > byte.MaxValue ? 2 : 1); + int paddingCountSize = GetPaddingCountSize(); int maxPayloadSize = maxPlainTextSize - @@ -618,9 +731,8 @@ namespace Opc.Ua.Bindings chunksToProcess.Add(new ArraySegment(buffer, headerSize, 0)); } - var chunksToSend = new BufferCollection(chunksToProcess.Capacity); - int messageSize = 0; + int keptCount = chunksToProcess.Count; for (int ii = 0; ii < chunksToProcess.Count; ii++) { @@ -635,10 +747,6 @@ namespace Opc.Ua.Bindings continue; } - // The headers are fixed-size little-endian fields in front of the - // body, written directly rather than through an encoder per chunk. - uint chunkType; - // check if the message needs to be aborted. The segment holds // only the body (it starts after the headers), which is what // the receiver counts against MaxMessageSize. @@ -647,8 +755,6 @@ namespace Opc.Ua.Bindings messageSize + chunkToProcess.Count, ii + 1)) { - chunkType = messageType | TcpMessageType.Abort; - // Replace the body in the chunk with an error message. // The encoder is bounded by the room the chunk has for a // body, not by what the chunk currently holds: an error @@ -667,24 +773,118 @@ namespace Opc.Ua.Bindings ? StatusCodes.BadRequestTooLarge : StatusCodes.BadResponseTooLarge); int size = errorEncoder.Close(); - chunkToProcess = new ArraySegment( + chunksToProcess[ii] = new ArraySegment( chunkArray, chunkToProcess.Offset, size); } limitsExceeded = true; + keptCount = ii + 1; } - // check if the message is complete. - else if (ii == chunksToProcess.Count - 1) + // update message size count. + messageSize += chunksToProcess[ii].Count; + } + + // the chunks after an abort chunk were returned above. + if (keptCount < chunksToProcess.Count) + { + chunksToProcess.RemoveRange(keptCount, chunksToProcess.Count - keptCount); + } + + // A small message that will wait in the write queue (writes are already + // queued) moves to a buffer of its own size: a deep pipeline of + // responses that each hold a send-buffer-sized array exhausts the + // buffer pool. A message that is sent right away skips the copy. The + // room after the body covers the padding count, the padding and the + // signature that sealing adds. + if (chunksToProcess.Count == 1 && Volatile.Read(ref m_activeWriteRequests) > 0) + { + ArraySegment body = chunksToProcess[0]; + int required = headerSize + body.Count + paddingCountSize + + signatureSize + (2 * EncryptionBlockSize); + if (required <= sendBufferSize / kSmallMessageFraction) { - chunkType = messageType | TcpMessageType.Final; + byte[] large = body.GetArray(); + byte[] small = BufferManager.TakeBuffer(required, "WriteSymmetricMessage"); + Buffer.BlockCopy(large, body.Offset, small, headerSize, body.Count); + chunksToProcess[0] = new ArraySegment(small, headerSize, body.Count); + BufferManager.ReturnBuffer(large, "WriteSymmetricMessage"); + } + } + + var encoded = new EncodedSymmetricMessage( + messageType, + requestId, + isRequest, + limitsExceeded, + chunksToProcess); + + // ensure the buffers don't get cleaned up on exit. + success = true; + return encoded; + } + finally + { + if (!success) + { + chunksToProcess?.Release(BufferManager, "WriteSymmetricMessage"); + } + } + } + + /// + /// Writes the headers and sequence numbers of an encoded message and + /// secures its chunks with the token. + /// + /// The message. Its chunks are consumed: returned + /// as the result, or released when sealing fails. + /// The token that secures the message. + /// The chunks to send. + /// + /// The caller holds , which is what keeps the + /// sequence numbers in the order the chunks reach the wire and keeps the + /// token from being disposed while it is in use. + /// + private protected BufferCollection SealSymmetricMessage( + EncodedSymmetricMessage encoded, + ChannelToken token) + { + BufferCollection chunksToProcess = encoded.Chunks; + bool success = false; + + try + { + int signatureSize = SymmetricSignatureSize; + int paddingCountSize = GetPaddingCountSize(); + + const int headerSize = TcpMessageLimits.SymmetricHeaderSize + + TcpMessageLimits.SequenceHeaderSize; + + var chunksToSend = new BufferCollection(chunksToProcess.Count); + int lastChunk = chunksToProcess.Count - 1; + + for (int ii = 0; ii < chunksToProcess.Count; ii++) + { + ArraySegment chunkToProcess = chunksToProcess[ii]; + // BufferManager-backed segment always has a non-null backing array. + byte[] chunkArray = chunkToProcess.GetArray(); + + // The headers are fixed-size little-endian fields in front of the + // body, written directly rather than through an encoder per chunk. + uint chunkType; + if (ii < lastChunk) + { + chunkType = encoded.MessageType | TcpMessageType.Intermediate; + } + else if (encoded.Aborted) + { + chunkType = encoded.MessageType | TcpMessageType.Abort; } - // more chunks to follow. else { - chunkType = messageType | TcpMessageType.Intermediate; + chunkType = encoded.MessageType | TcpMessageType.Final; } int count = 0; @@ -721,14 +921,11 @@ namespace Opc.Ua.Bindings uint sequenceNumber = GetNewSequenceNumber(); BinaryPrimitives.WriteUInt32LittleEndian(header[16..], sequenceNumber); - BinaryPrimitives.WriteUInt32LittleEndian(header[20..], requestId); + BinaryPrimitives.WriteUInt32LittleEndian(header[20..], encoded.RequestId); // the headers are followed by the body. int position = headerSize + chunkToProcess.Count; - // update message size count. - messageSize += chunkToProcess.Count; - ArraySegment dataToSend; if (SecurityMode != MessageSecurityMode.None) @@ -738,7 +935,13 @@ namespace Opc.Ua.Bindings TcpMessageLimits.SymmetricHeaderSize, position - TcpMessageLimits.SymmetricHeaderSize); - dataToSend = EncryptAndSign(token, dataToSend, isRequest); + // the number was taken above under the same lock, so the + // last one sent is the one before this chunk's. + dataToSend = EncryptAndSign( + token, + dataToSend, + encoded.IsRequest, + LastSentSequenceNumber); } else { @@ -760,15 +963,27 @@ namespace Opc.Ua.Bindings { if (!success) { - chunksToProcess?.Release(BufferManager, "WriteSymmetricMessage"); + chunksToProcess.Release(BufferManager, "WriteSymmetricMessage"); } } } + /// + /// The size of the padding count field of an encrypted chunk. + /// + private int GetPaddingCountSize() + { + return SecurityMode != MessageSecurityMode.SignAndEncrypt || + SecurityPolicy!.NoSymmetricEncryptionPadding + ? 0 + : (EncryptionBlockSize > byte.MaxValue ? 2 : 1); + } + private ArraySegment EncryptAndSign( ChannelToken token, ArraySegment dataToEncrypt, - bool useClientKeys) + bool useClientKeys, + uint lastSequenceNumber) { // The keys and security policy are populated in ComputeKeys before send/receive. return CryptoUtils.SymmetricEncryptAndSign( @@ -780,7 +995,7 @@ namespace Opc.Ua.Bindings useClientKeys ? token.ClientHmac : token.ServerHmac, SecurityMode == MessageSecurityMode.Sign, token.TokenId, - LastSentSequenceNumber, // already incremented to create this message. need the last one sent. + lastSequenceNumber, m_symmetricProvider, useClientKeys ? token.ClientCiphers : token.ServerCiphers); } @@ -978,6 +1193,12 @@ namespace Opc.Ua.Bindings /// was activated. /// private ChannelToken? m_renewedTokenWithKeys; + + /// + /// A single-chunk message that needs at most this fraction of the send buffer + /// is moved to a buffer of its own size (see ). + /// + private const int kSmallMessageFraction = 8; private int m_signatureKeySize; private int m_encryptionKeySize; private ISymmetricCryptoProvider? m_symmetricProvider; diff --git a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs index dcaaab9ee3..e0bea36d7d 100644 --- a/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs +++ b/src/Opc.Ua.Core/Stack/Tcp/UaSCBinaryChannel.cs @@ -1449,9 +1449,10 @@ namespace Opc.Ua.Bindings /// Both matter. Running off the caller's stack is required because the /// write reports completion through , /// which enters the gate the caller may be holding. Preserving order is - /// required because sequence numbers are assigned under that gate, and a - /// peer rejects a chunk whose sequence number does not follow its - /// predecessor. Independently queued work items carry no such guarantee, + /// required because sequence numbers are assigned in queue order (see + /// ), and a peer rejects a chunk whose sequence + /// number does not follow its predecessor. Independently queued work items + /// carry no such guarantee, /// so the writes are queued and a single writer on the thread pool sends /// them one after the other; a write cannot start before the one issued /// before it has finished. @@ -1468,6 +1469,61 @@ namespace Opc.Ua.Bindings m_writerRunning = true; } + ScheduleWriter(); + } + + /// + /// Queues a message encoded by for the + /// writer, which seals it in queue order and sends it. + /// + /// The message. Ownership transfers when this + /// returns . + /// Reported to . + /// when no transport is attached; the + /// caller still owns the message. + /// + /// The caller needs no lock: nothing ordered per channel is decided until + /// the writer seals the message. See . + /// + /// The writer always starts on the pool, never on the caller's thread. Run + /// on the thread that produced the first response, it sealed and sent + /// batches of one message, which cost more send calls and serialised the + /// signing on the producers; on four cores, encrypted, it lost 30 % of the + /// throughput a pool-scheduled writer reaches with larger coalesced batches. + /// + /// + private protected bool BeginWriteSymmetricMessage( + EncodedSymmetricMessage message, + object? state) + { + IUaSCByteTransport? transport = m_transport; + if (transport == null) + { + return false; + } + + Interlocked.Increment(ref m_activeWriteRequests); + var write = new PendingWrite(transport, default, null, null, state, message); + + lock (m_writeQueueLock) + { + m_unsealedWriteQueue.Enqueue(write); + if (m_writerRunning) + { + return true; + } + m_writerRunning = true; + } + + ScheduleWriter(); + return true; + } + + /// + /// Starts the writer on the thread pool. + /// + private void ScheduleWriter() + { // The execution context is not needed by the write and is not flowed. // The write goes to the local queue of the issuing thread, which picks // it up as soon as it returns to the pool (typically to await the @@ -1488,46 +1544,70 @@ namespace Opc.Ua.Bindings #endif /// - /// Sends the queued writes in order until the queue is empty. + /// Seals and sends the queued writes in order until the queue is empty. /// private async Task RunWritesAsync() { List batch = m_writeBatch; while (true) { - PendingWrite write; + bool numbered; lock (m_writeQueueLock) { - if (m_writeQueue.Count == 0) + if (m_writeQueue.Count == 0 && m_unsealedWriteQueue.Count == 0) { m_writerRunning = false; 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()); - } - } + // Writes numbered by their producers go out without the send + // order lock: each was queued under it, after the writes numbered + // before it. A client channel only queues such writes. + numbered = m_writeQueue.Count > 0 && TakeWriteBatch(batch, includeUnsealed: false); } - Task pending; - if (batch.Count > 0) + if (numbered) { - pending = WriteCoalescedAsync(write.Transport, batch); + DiscardWritesNumberedAfterGap(batch); } else { + // Sealing takes sequence numbers and the current token, so it runs + // under the send order lock, and in queue order. The batch is taken + // under the same lock: see SendOrder for why that orders the wire. + using (await SendOrder.EnterAsync().ConfigureAwait(false)) + { + lock (m_writeQueueLock) + { + if (!TakeWriteBatch(batch, includeUnsealed: true)) + { + m_writerRunning = false; + return; + } + } + + DiscardWritesNumberedAfterGap(batch); + SealWriteBatch(batch); + } + + ReportSequenceFailure(); + } + + if (batch.Count == 0) + { + // every message in the batch failed to seal and was reported. + continue; + } + + Task pending; + if (batch.Count > 1) + { + pending = WriteCoalescedAsync(batch[0].Transport, batch); + } + else + { + PendingWrite write = batch[0]; + batch.Clear(); pending = write.Buffers != null ? WriteBuffersAsync(write.Transport, write.Buffers, write.State) : WriteSingleChunkAsync(write.Transport, write.Chunk, write.BackingBuffer!, write.State); @@ -1555,9 +1635,219 @@ namespace Opc.Ua.Bindings } } + /// + /// Takes the next writes to send together, the numbered ones first. + /// + /// Receives the writes; empty on entry. + /// Whether unsealed writes may be taken; only + /// a caller that holds may number them. + /// when nothing is queued that may be taken. + /// + /// Called under the write queue lock, and under when + /// is set. Every sealed write in the + /// queue was numbered before the caller took , so it + /// goes out before the unsealed writes the caller is about to number. Writes + /// queued behind the first one for the same transport go out with it in one + /// vectored send, when the transport is a stream. + /// + private bool TakeWriteBatch(List batch, bool includeUnsealed) + { + PendingWrite first; + if (m_writeQueue.Count > 0) + { + first = m_writeQueue.Dequeue(); + } + else if (includeUnsealed && m_unsealedWriteQueue.Count > 0) + { + first = m_unsealedWriteQueue.Dequeue(); + } + else + { + return false; + } + + batch.Add(first); + + if (first.Transport is IUaSCChunkCoalescing coalescing && + coalescing.CanCoalesceChunks) + { + while (m_writeQueue.Count > 0 && + batch.Count < kMaxCoalescedWrites && + ReferenceEquals(m_writeQueue.Peek().Transport, first.Transport)) + { + batch.Add(m_writeQueue.Dequeue()); + } + + // unsealed writes only once no sealed write is left ahead of them. + while (includeUnsealed && + m_writeQueue.Count == 0 && + m_unsealedWriteQueue.Count > 0 && + batch.Count < kMaxCoalescedWrites && + ReferenceEquals(m_unsealedWriteQueue.Peek().Transport, first.Transport)) + { + batch.Add(m_unsealedWriteQueue.Dequeue()); + } + } + + return true; + } + + /// + /// Seals the unsealed writes of a batch in order, and drops and reports + /// those that cannot be sealed. + /// + /// + /// Called under . A write that fails is reported as + /// a failed write, after its chunks have been returned. + /// + /// A message that fails after it took sequence numbers leaves a gap the + /// peer rejects. The writer then numbers nothing more for that transport, + /// and faults the channel once it has left + /// the lock. + /// + /// + private void SealWriteBatch(List batch) + { + int kept = 0; + for (int ii = 0; ii < batch.Count; ii++) + { + PendingWrite write = batch[ii]; + if (write.Unsealed is EncodedSymmetricMessage message) + { + // A message queued for a transport the channel no longer uses (a + // reconnect replaced it, or the channel closed) is not numbered: + // after a reconnect the peer continues from the OpenSecureChannel + // response on the new transport and would see a gap. It is + // dropped like a write to the closed transport. + // After a message failed once it had taken sequence numbers, nothing + // more is numbered for that transport either: the peer would see the gap. + ChannelToken? token = CurrentToken; + if (token == null || + !ReferenceEquals(write.Transport, m_transport) || + ReferenceEquals(write.Transport, m_sequenceBrokenTransport)) + { + message.Chunks.Release(BufferManager, nameof(SealWriteBatch)); + ReportWriteComplete( + null, + write.State, + 0, + ServiceResult.Create( + StatusCodes.BadSecureChannelClosed, + "The channel was closed before the message was sent.")); + continue; + } + + long numbered = Interlocked.Read(ref m_sequenceNumber); + try + { + // returns the chunks to the pool when it fails. + BufferCollection sealedChunks = SealSymmetricMessage(message, token); + write = new PendingWrite(write.Transport, default, null, sealedChunks, write.State); + } + catch (Exception ex) + { + m_logger.UaSCChannelSealFailed(ex, ChannelId, message.RequestId); + var failure = ServiceResult.Create( + ex, + StatusCodes.BadTcpInternalError, + "Could not secure the outgoing message."); + if (Interlocked.Read(ref m_sequenceNumber) != numbered) + { + // it failed after taking sequence numbers (its security failed). + m_sequenceBrokenTransport = write.Transport; + m_sequenceFailure = failure; + } + ReportWriteComplete(null, write.State, 0, failure); + continue; + } + } + + batch[kept++] = write; + } + + if (kept < batch.Count) + { + batch.RemoveRange(kept, batch.Count - kept); + } + } + + /// + /// Drops the writes of a batch that their producers numbered for a transport + /// on which a failed seal already left a gap in the sequence numbers. + /// + /// + /// A producer can number a message for that transport (an OpenSecureChannel + /// response or a fault, or a client request) before the channel faults: + /// faults it from the pool. Sent, it + /// would reach the peer after the gap. The writer seals only once no write + /// numbered by a producer is queued ahead, and producers cannot number while + /// it holds , so every such write queued for the + /// broken transport was numbered after the gap. Unsealed writes are left to + /// . + /// + private void DiscardWritesNumberedAfterGap(List batch) + { + IUaSCByteTransport? broken = m_sequenceBrokenTransport; + if (broken == null) + { + return; + } + + int kept = 0; + for (int ii = 0; ii < batch.Count; ii++) + { + PendingWrite write = batch[ii]; + if (write.Unsealed == null && ReferenceEquals(write.Transport, broken)) + { + if (write.Buffers == null && write.BackingBuffer != null) + { + BufferManager.ReturnBuffer(write.BackingBuffer, nameof(DiscardWritesNumberedAfterGap)); + } + ReportWriteComplete( + write.Buffers, + write.State, + 0, + ServiceResult.Create( + StatusCodes.BadSecureChannelClosed, + "The channel was closed before the message was sent.")); + continue; + } + + batch[kept++] = write; + } + + if (kept < batch.Count) + { + batch.RemoveRange(kept, batch.Count - kept); + } + } + + /// + /// Faults the channel on the transport whose sequence numbers a failed seal + /// left with a gap, once the writer no longer holds . + /// + /// + /// The fault takes the gate, which comes before in + /// the lock order, so it runs on the pool, as a failed write is reported. + /// It is ignored once a reconnect replaced the transport. + /// + private void ReportSequenceFailure() + { + ServiceResult? failure = m_sequenceFailure; + if (failure == null) + { + return; + } + + m_sequenceFailure = null; + IUaSCByteTransport transport = m_sequenceBrokenTransport!; + _ = Task.Run(() => OnTransportError(transport, failure, CancellationToken.None)); + } + /// /// A write waiting for its turn: either one contiguous chunk with the - /// buffer to return, or a collection of buffers. + /// buffer to return, a collection of buffers, or a message the writer + /// still has to seal. /// private readonly struct PendingWrite { @@ -1566,13 +1856,15 @@ namespace Opc.Ua.Bindings ReadOnlyMemory chunk, byte[]? backingBuffer, BufferCollection? buffers, - object? state) + object? state, + EncodedSymmetricMessage? unsealed = null) { Transport = transport; Chunk = chunk; BackingBuffer = backingBuffer; Buffers = buffers; State = state; + Unsealed = unsealed; } public IUaSCByteTransport Transport { get; } @@ -1580,6 +1872,7 @@ namespace Opc.Ua.Bindings public byte[]? BackingBuffer { get; } public BufferCollection? Buffers { get; } public object? State { get; } + public EncodedSymmetricMessage? Unsealed { get; } } /// @@ -2070,6 +2363,41 @@ namespace Opc.Ua.Bindings /// internal ChannelGate Gate { get; } = new(); + /// + /// Orders the sequence numbers of outgoing chunks with the order they + /// reach the transport, and guards the token the writer secures them with. + /// + /// + /// OPC 10000-6 §6.7.2.4 requires the sequence numbers on the wire to + /// increase by one per chunk, and a message's chunks to be contiguous. + /// Messages are numbered in one of two ways, both under this lock: + /// + /// The writer seals unsealed messages + /// () while it holds the lock, + /// in queue order, after taking them from the queue under it. + /// A producer that numbers its own message (asymmetric messages, + /// ) holds the lock from taking the + /// numbers until the message is in the queue of sealed writes, unless its + /// channel queues no unsealed messages and orders those producers itself + /// (the client, under its gate). + /// + /// The writer sends every sealed write ahead of the unsealed ones it takes + /// with them, so a write numbered by a producer — necessarily before the + /// writer took the lock — precedes the numbers the writer assigns, and one + /// writer sends batches one after the other. Sealed writes alone the writer + /// takes without the lock: they are in the queue in the order of their + /// numbers, and a write still being numbered under the lock is queued after + /// them. Messages without sequence numbers (Hello, Acknowledge, + /// ReverseHello, Error) need no lock. + /// + /// Token changes that can dispose a token take the lock too, so the writer + /// never secures a chunk with a disposed token. Lock order: , + /// then this, then the write queue lock. Nothing takes the gate while + /// holding this. + /// + /// + internal ChannelGate SendOrder { get; } = new(); + /// /// The byte-level transport that carries UASC chunks for the channel. /// Set by listener channels after a successful accept/connect, by @@ -2274,8 +2602,21 @@ namespace Opc.Ua.Bindings private TaskCompletionSource? m_writeCapacityWaiter; private readonly Lock m_writeQueueLock = new(); private readonly Queue m_writeQueue = new(); + private readonly Queue m_unsealedWriteQueue = new(); private bool m_writerRunning; + /// + /// The transport on which a message failed after taking sequence numbers; + /// nothing more is numbered for it. Changed under . + /// + private IUaSCByteTransport? m_sequenceBrokenTransport; + + /// + /// Why was set, until the writer + /// reports it (). + /// + private ServiceResult? m_sequenceFailure; + /// /// The most writes sent together in one vectored send. /// @@ -2626,5 +2967,13 @@ namespace Opc.Ua.Bindings uint channelId, long limitBytes, bool hasSession); + + [LoggerMessage(EventId = CoreEventIds.UaSCBinaryChannel + 18, Level = LogLevel.Warning, + Message = "ChannelId {ChannelId}: Could not secure the message for RequestId {RequestId}; it is not sent.")] + public static partial void UaSCChannelSealFailed( + this ILogger logger, + Exception exception, + uint channelId, + uint requestId); } } diff --git a/src/Opc.Ua.Types/Utils/ServiceMessageContext.cs b/src/Opc.Ua.Types/Utils/ServiceMessageContext.cs index 715267c089..1a754235ae 100644 --- a/src/Opc.Ua.Types/Utils/ServiceMessageContext.cs +++ b/src/Opc.Ua.Types/Utils/ServiceMessageContext.cs @@ -147,48 +147,49 @@ namespace Opc.Ua public int MaxDecoderRecoveries { get; set; } /// + /// + /// Read without a lock: every encoder and decoder of a server reads it, from + /// all request threads at once, and a lock around a reference read became a + /// waiting point once responses were encoded concurrently. The tables and + /// are published together, as one immutable + /// snapshot, so a reader that sees the version unchanged around reading a + /// table read the table of that version. Setters serialise on a lock. + /// public NamespaceTable NamespaceUris { - get - { - lock (m_mappingGate) - { - return m_namespaceUris; - } - } + get => Volatile.Read(ref m_mapping).NamespaceUris; set { lock (m_mappingGate) { value ??= new NamespaceTable(); - if (!ReferenceEquals(m_namespaceUris, value)) + MappingSnapshot current = m_mapping; + if (!ReferenceEquals(current.NamespaceUris, value)) { - m_namespaceUris = value; - m_mappingVersion++; + Volatile.Write( + ref m_mapping, + new MappingSnapshot(value, current.ServerUris, current.Version + 1)); } } } } /// + /// Read without a lock; see . public StringTable ServerUris { - get - { - lock (m_mappingGate) - { - return m_serverUris; - } - } + get => Volatile.Read(ref m_mapping).ServerUris; set { lock (m_mappingGate) { value ??= new StringTable(); - if (!ReferenceEquals(m_serverUris, value)) + MappingSnapshot current = m_mapping; + if (!ReferenceEquals(current.ServerUris, value)) { - m_serverUris = value; - m_mappingVersion++; + Volatile.Write( + ref m_mapping, + new MappingSnapshot(current.NamespaceUris, value, current.Version + 1)); } } } @@ -215,20 +216,28 @@ namespace Opc.Ua /// /// Gets the monotonic version of the namespace/server table holders. /// - internal long MappingVersion + internal long MappingVersion => Volatile.Read(ref m_mapping).Version; + + /// + /// The namespace and server tables with the version they were published at. + /// + private sealed class MappingSnapshot { - get + public MappingSnapshot(NamespaceTable namespaceUris, StringTable serverUris, long version) { - lock (m_mappingGate) - { - return m_mappingVersion; - } + NamespaceUris = namespaceUris; + ServerUris = serverUris; + Version = version; } + + public NamespaceTable NamespaceUris { get; } + + public StringTable ServerUris { get; } + + public long Version { get; } } private readonly Lock m_mappingGate = new(); - private NamespaceTable m_namespaceUris = new(); - private StringTable m_serverUris = new(); - private long m_mappingVersion; + private MappingSnapshot m_mapping = new(new NamespaceTable(), new StringTable(), 0); } } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs index 9ef9f099c1..5c3c56a4e5 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpByteTransportTests.cs @@ -294,10 +294,10 @@ namespace Opc.Ua.Core.Tests.Stack.Transport using Socket _s = serverSocket; using (client) { - var sendLock = (SemaphoreSlim)typeof(TcpByteTransport) + var sendLock = (ChannelGate)typeof(TcpByteTransport) .GetField("m_sendLock", BindingFlags.Instance | BindingFlags.NonPublic)! .GetValue(client)!; - sendLock.Wait(); + ChannelGate.Releaser held = sendLock.Enter(); try { using var cts = new CancellationTokenSource(); @@ -314,7 +314,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport } finally { - sendLock.Release(); + held.Dispose(); } } } @@ -328,10 +328,10 @@ namespace Opc.Ua.Core.Tests.Stack.Transport using Socket _s = serverSocket; using (client) { - var sendLock = (SemaphoreSlim)typeof(TcpByteTransport) + var sendLock = (ChannelGate)typeof(TcpByteTransport) .GetField("m_sendLock", BindingFlags.Instance | BindingFlags.NonPublic)! .GetValue(client)!; - sendLock.Wait(); + ChannelGate.Releaser held = sendLock.Enter(); try { Task send = client.SendChunkAsync( @@ -349,7 +349,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport } finally { - sendLock.Release(); + held.Dispose(); } } } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpReconnectKeyAgreementTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpReconnectKeyAgreementTests.cs index c4f80236c6..f9647544bf 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpReconnectKeyAgreementTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpReconnectKeyAgreementTests.cs @@ -235,6 +235,63 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That(harness.Target.Token.Secret, Is.EqualTo(harness.Peer.Token.Secret)); } + /// + /// A response still queued for the dropped transport when a reconnect activates the + /// new token is not secured with that token and sent: the reconnect detaches the old + /// transport first, so the writer drops it without numbering it, and the + /// OpenSecureChannel response is the first message on the new transport. + /// + [TestCase(SecurityPolicies.ECC_nistP256)] + [TestCase(SecurityPolicies.RSA_DH_AesGcm)] + public async Task AResponseQueuedForTheDroppedTransportIsNotSecuredWithTheNewTokenAsync(string policyUri) + { + if (!AssertPolicyAvailability(policyUri)) + { + return; + } + using var harness = new HandoffHarness(policyUri); + await harness.OpenAsync().ConfigureAwait(false); + int sentOnOld = harness.OldTransport.SentCount; + + // the writer sends one response on the old transport, and the next waits behind it. + harness.OldTransport.HoldSends(); + harness.Target.SendResponseForTest(70, new ReadResponse()); + Assert.That( + SpinWait.SpinUntil(() => harness.OldTransport.SentCount > sentOnOld, TimeSpan.FromSeconds(10)), + Is.True); + harness.Target.SendResponseForTest(71, new ReadResponse()); + + // the writer goes on once the reconnect has activated the new token. + Action? activated = harness.Target.TokenActivatedCallback; + bool armed = true; + harness.Target.TokenActivatedCallback = (current, previous) => + { + activated?.Invoke(current, previous); + if (armed) + { + armed = false; + harness.OldTransport.ReleaseSends(); + Assert.That( + SpinWait.SpinUntil(() => !harness.Target.HasPendingWritesForTest, TimeSpan.FromSeconds(10)), + Is.True); + } + }; + + await harness.RenewAsync().ConfigureAwait(false); + + Assert.That(harness.HandoffError, Is.Null); + Assert.That(armed, Is.False, "the token was not activated."); + Assert.That( + harness.OldTransport.SentCount, + Is.EqualTo(sentOnOld + 1), + "the response queued for the dropped transport was sent with the new token."); + await harness.Peer.CompleteOpenAsync( + await harness.NewTransport.ReadAsync(harness.CancellationToken).ConfigureAwait(false), + renew: true, + reconnect: true).ConfigureAwait(false); + await harness.AssertEncryptedReadAsync(harness.NewTransport).ConfigureAwait(false); + } + [TestCase(SecurityPolicies.ECC_nistP256, false)] [TestCase(SecurityPolicies.ECC_nistP256, true)] [TestCase(SecurityPolicies.RSA_DH_AesGcm, false)] @@ -999,6 +1056,18 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That(VerifySequenceNumber(sequenceNumber, "ReceivedBeforeReconnect"), Is.True); } + /// + /// Queues a response for the writer on the transport the channel has now. + /// + public void SendResponseForTest(uint requestId, IServiceResponse response) + { + EncodedSymmetricMessage encoded = EncodeSymmetricMessage( + TcpMessageType.Message, requestId, response, false, out _); + Assert.That(BeginWriteSymmetricMessage(encoded, null), Is.True); + } + + public bool HasPendingWritesForTest => HasPendingWrites; + protected internal override void StartReceiveLoop() { if (FailReceiveStart) @@ -1146,10 +1215,29 @@ namespace Opc.Ua.Core.Tests.Stack.Transport throw new NotSupportedException(); } - public ValueTask SendChunkAsync(ReadOnlyMemory chunk, CancellationToken ct) + public async ValueTask SendChunkAsync(ReadOnlyMemory chunk, CancellationToken ct) { Interlocked.Increment(ref m_sentCount); - return m_sent.Writer.WriteAsync(chunk.ToArray().ToByteString(), ct); + ByteString sent = chunk.ToArray().ToByteString(); + TaskCompletionSource? hold = Volatile.Read(ref m_hold); + if (hold != null) + { + await hold.Task.ConfigureAwait(false); + } + await m_sent.Writer.WriteAsync(sent, ct).ConfigureAwait(false); + } + + /// + /// Holds every send that starts from now on until . + /// + public void HoldSends() + { + Volatile.Write(ref m_hold, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + } + + public void ReleaseSends() + { + Interlocked.Exchange(ref m_hold, null)?.TrySetResult(true); } public async ValueTask SendChunkAsync(BufferCollection buffers, CancellationToken ct) @@ -1187,6 +1275,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport } private readonly Channel m_sent = Channel.CreateUnbounded(); + private TaskCompletionSource? m_hold; private int m_closed; private int m_sentCount; } diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpServerChannelBufferTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpServerChannelBufferTests.cs index 8f4f263291..869bfe6785 100644 --- a/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpServerChannelBufferTests.cs +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/TcpServerChannelBufferTests.cs @@ -62,9 +62,11 @@ namespace Opc.Ua.Core.Tests.Stack.Transport var transport = new GateByteTransport(responseCount); channel.SetTransport(transport); + // responses above an eighth of the send buffer keep their send buffer; + // smaller ones move to a buffer of their own size. for (uint requestId = 1; requestId <= responseCount; requestId++) { - channel.SendResponse(requestId, CreateResponse()); + channel.SendResponse(requestId, CreateLargeResponse(16 * 1024)); } Assert.That( @@ -103,8 +105,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That( await WaitForOutstandingCountAsync(pool, expected: 0, 30).ConfigureAwait(false), Is.True); - Assert.That(pool.RentCount, Is.EqualTo(1)); - Assert.That(pool.ReturnCount, Is.EqualTo(1)); + Assert.That(pool.ReturnCount, Is.EqualTo(pool.RentCount)); Assert.That(pool.DuplicateReturnCount, Is.Zero); } @@ -223,8 +224,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That( await WaitForOutstandingCountAsync(pool, expected: 0, 30).ConfigureAwait(false), Is.True); - Assert.That(pool.RentCount, Is.EqualTo(1)); - Assert.That(pool.ReturnCount, Is.EqualTo(1)); + Assert.That(pool.ReturnCount, Is.EqualTo(pool.RentCount)); Assert.That(pool.DuplicateReturnCount, Is.Zero); } @@ -248,8 +248,7 @@ namespace Opc.Ua.Core.Tests.Stack.Transport Assert.That( await WaitForOutstandingCountAsync(pool, expected: 0, 30).ConfigureAwait(false), Is.True); - Assert.That(pool.RentCount, Is.EqualTo(1)); - Assert.That(pool.ReturnCount, Is.EqualTo(1)); + Assert.That(pool.ReturnCount, Is.EqualTo(pool.RentCount)); Assert.That(pool.DuplicateReturnCount, Is.Zero); } @@ -337,7 +336,9 @@ namespace Opc.Ua.Core.Tests.Stack.Transport var transport = new GateByteTransport(expectedSendCount: 1); channel.SetTransport(transport); - channel.SendResponse(1, CreateResponse()); + // large enough to keep the send buffer (smaller responses move to a + // buffer of their own size). + channel.SendResponse(1, CreateLargeResponse(16 * 1024)); Assert.That( await CompletesWithinAsync(transport.AllSendsStarted, 30).ConfigureAwait(false), diff --git a/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelSendOrderTests.cs b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelSendOrderTests.cs new file mode 100644 index 0000000000..d49b664ba2 --- /dev/null +++ b/tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelSendOrderTests.cs @@ -0,0 +1,1267 @@ +/* ======================================================================== + * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. + * + * OPC Foundation MIT License 1.00 + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following + * conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + * OTHER DEALINGS IN THE SOFTWARE. + * + * The complete license agreement can be found here: + * http://opcfoundation.org/License/MIT/1.00/ + * ======================================================================*/ + +using System; +using System.Buffers; +using System.Buffers.Binary; +using System.Collections.Generic; +using System.Linq; +using System.Net; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Time.Testing; +using Moq; +using NUnit.Framework; +using Opc.Ua.Bindings; +using Opc.Ua.Security.Certificates; +using Opc.Ua.Tests; + +// CA5394: the random choices only vary the interleaving of the tests; the seeds are fixed or logged. +#pragma warning disable CA5394 + +namespace Opc.Ua.Core.Tests.Stack.Transport +{ + /// + /// Responses are encoded on the producer's thread without the channel gate and + /// numbered and secured by the channel's single writer (see + /// UaSCUaBinaryChannel.SendOrder). These tests check what that must keep: + /// sequence numbers that increase by one in wire order, contiguous chunks, the + /// token rules across a renewal, and buffers returned exactly once. + /// + [TestFixture] + [Category("TransportChannelDeterministic")] + [SetCulture("en-us")] + [SetUICulture("en-us")] + [NonParallelizable] + public sealed class UaSCBinaryChannelSendOrderTests + { + [Test] + public async Task ConcurrentResponsesReachTheWireInSequenceOrderAsync() + { + const int producers = 8; + const int perProducer = 150; + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 3); + channel.SetTransport(transport); + + await RunProducersAsync(producers, async (producer, random) => + { + for (int ii = 0; ii < perProducer; ii++) + { + uint requestId = (uint)((producer * perProducer) + ii + 1); + bool retained = await channel + .SendResponseAsync(requestId, CreateResponse(random)) + .ConfigureAwait(false); + Assert.That(retained, Is.False); + if (random.Next(4) == 0) + { + await Task.Yield(); + } + } + }).ConfigureAwait(false); + + List chunks = await transport + .WaitForFinalChunksAsync(producers * perProducer) + .ConfigureAwait(false); + AssertWireOrder(chunks, Enumerable.Range(1, producers * perProducer).Select(id => (uint)id)); + Assert.That(chunks.Any(chunk => !chunk.IsFinal), Is.True, "no response was split into chunks."); + + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + + [Test] + public async Task MessagesNumberedUnderSendOrderKeepTheWireOrderAsync() + { + // An asymmetric message (an OpenSecureChannel response to a renewal) takes + // its sequence number when it is built, while responses are numbered by + // the writer. Both kinds sent concurrently must still be in order. + const int producers = 6; + const int perProducer = 120; + const int numbered = 200; + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 2); + channel.SetTransport(transport); + + await RunProducersAsync(producers + 1, async (producer, random) => + { + if (producer == producers) + { + for (int ii = 0; ii < numbered; ii++) + { + channel.QueueNumberedMessageForTest((uint)(100_000 + ii)); + if (random.Next(3) == 0) + { + await Task.Yield(); + } + } + return; + } + + for (int ii = 0; ii < perProducer; ii++) + { + uint requestId = (uint)((producer * perProducer) + ii + 1); + await channel.SendResponseAsync(requestId, CreateResponse(random)).ConfigureAwait(false); + } + }).ConfigureAwait(false); + + IEnumerable expected = Enumerable.Range(1, producers * perProducer).Select(id => (uint)id) + .Concat(Enumerable.Range(100_000, numbered).Select(id => (uint)id)); + List chunks = await transport + .WaitForFinalChunksAsync((producers * perProducer) + numbered) + .ConfigureAwait(false); + AssertWireOrder(chunks, expected); + + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + + [TestCase(SecurityPolicies.Basic256Sha256, MessageSecurityMode.SignAndEncrypt)] + [TestCase(SecurityPolicies.Basic256Sha256, MessageSecurityMode.Sign)] + [TestCase(SecurityPolicies.Aes256_Sha256_RsaPss, MessageSecurityMode.SignAndEncrypt)] + [TestCase(SecurityPolicies.ECC_nistP256, MessageSecurityMode.SignAndEncrypt)] + [TestCase(SecurityPolicies.ECC_nistP256_AesGcm, MessageSecurityMode.SignAndEncrypt)] + public async Task TokenRenewalDuringConcurrentSendsKeepsMessagesVerifiableAsync( + string policyUri, + MessageSecurityMode mode) + { + SecurityPolicyInfo? policy = SecurityPolicies.Default.GetInfo(policyUri); + if (policy == null) + { + Assert.Ignore("The policy is not supported on this platform."); + return; + } + + const int producers = 4; + const int perProducer = 100; + const int renewals = 12; + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var pool = new CountingPool(); + using SealProbe writer = SealProbe.Create(policyUri, mode, telemetry, pool); + using SealProbe reader = SealProbe.Create(policyUri, mode, telemetry, new CountingPool()); + var transport = new RecordingTransport(asyncSendEvery: 2); + writer.SetTransport(transport); + + var readerTokens = new Dictionary(); + try + { + SealProbe.Renew(writer, reader, 1, readerTokens); + + // the OpenSecureChannel exchange consumed the first sequence number; + // the AEAD policies feed the previous one into the nonce. + Assert.That(reader.AcceptSequenceNumber(writer.TakeSequenceNumber()), Is.True); + + using var sendsDone = new CancellationTokenSource(); + Task renewer = Task.Run(async () => + { + for (uint tokenId = 2; tokenId <= renewals + 1; tokenId++) + { + await Task.Delay(2).ConfigureAwait(false); + lock (readerTokens) + { + SealProbe.Renew(writer, reader, tokenId, readerTokens); + } + } + }); + + await RunProducersAsync(producers, async (producer, random) => + { + for (int ii = 0; ii < perProducer; ii++) + { + uint requestId = (uint)((producer * perProducer) + ii + 1); + writer.Send(requestId, CreateResponse(random)); + if (random.Next(3) == 0) + { + await Task.Delay(random.Next(2)).ConfigureAwait(false); + } + } + }).ConfigureAwait(false); + await renewer.ConfigureAwait(false); + + List chunks = await transport + .WaitForMessageCountAsync(producers * perProducer) + .ConfigureAwait(false); + + // decode in wire order, as the client does: every chunk verifies with the + // token it names, the numbers follow on, and once a newer token was used + // the older one is not used again. + uint lastTokenId = 0; + var decoded = new List(chunks.Count); + foreach (WireChunk chunk in chunks) + { + Assert.That(chunk.TokenId, Is.GreaterThanOrEqualTo(lastTokenId), "a token was used after its successor."); + lastTokenId = chunk.TokenId; + decoded.Add(reader.Read(chunk, readerTokens[chunk.TokenId])); + } + + Assert.That(lastTokenId, Is.GreaterThan(1u), "no message was secured with a renewed token."); + AssertWireOrder(decoded, Enumerable.Range(1, producers * perProducer).Select(id => (uint)id)); + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + finally + { + reader.ReleaseTokens(); + foreach (ChannelToken token in readerTokens.Values) + { + token.Dispose(); + } + } + } + + [Test] + public async Task ClosingWithUnsealedResponsesQueuedReturnsEveryBufferOnceAsync() + { + var pool = new CountingPool(); + OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 1); + transport.HoldSends(); + channel.SetTransport(transport); + + var random = new Random(7); + for (uint requestId = 1; requestId <= 40; requestId++) + { + await channel.SendResponseAsync(requestId, CreateResponse(random)).ConfigureAwait(false); + } + + // the first send is in flight; the rest wait in the queue unsealed. + await transport.FirstSendStarted.ConfigureAwait(false); + Assert.That(channel.HasQueuedWritesForTest, Is.True); + + // disposing discards the tokens; the writer finds no token to seal with. + channel.Dispose(); + transport.ReleaseSends(); + + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + Assert.That(channel.HasQueuedWritesForTest, Is.False); + } + + [Test] + public async Task SealFailureIsReportedAndLaterResponsesStillSendAsync() + { + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 1); + transport.HoldSends(); + channel.SetTransport(transport); + + await channel.SendResponseAsync(1, CreateResponse(new Random(1), 0)).ConfigureAwait(false); + await transport.FirstSendStarted.ConfigureAwait(false); + + // queued behind the held send, then the token goes away and comes back: + // the queued ones fail to seal, the ones after are sent. + await channel.SendResponseAsync(2, CreateResponse(new Random(2), 0)).ConfigureAwait(false); + await channel.SendResponseAsync(3, CreateResponse(new Random(3), 0)).ConfigureAwait(false); + channel.LoadTokenForTest(null); + transport.ReleaseSends(); + await WaitUntilAsync(() => !channel.HasQueuedWritesForTest).ConfigureAwait(false); + channel.LoadTokenForTest(2); + + await channel.SendResponseAsync(4, CreateResponse(new Random(4), 0)).ConfigureAwait(false); + + List chunks = await transport.WaitForFinalChunksAsync(2).ConfigureAwait(false); + Assert.That(chunks.Select(chunk => chunk.RequestId), Is.EqualTo(new uint[] { 1, 4 })); + Assert.That(chunks[1].SequenceNumber, Is.EqualTo(chunks[0].SequenceNumber + 1)); + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + + /// + /// A response that cannot be encoded is answered with a ServiceFault carrying + /// its RequestHandle, whether it is sent on the receive loop while that + /// dispatches the request (with the gate free) or from another thread. + /// + [TestCase(true)] + [TestCase(false)] + public async Task AResponseThatCannotBeEncodedIsAnsweredWithAFaultAsync(bool sentWhileDispatching) + { + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool, maxByteStringLength: 100); + var transport = new RecordingTransport(asyncSendEvery: 2); + channel.SetTransport(transport); + channel.SetRequestReceivedCallback((target, requestId, request) => + { + ReadResponse response = CreateResponse(null, requestId == 2 ? 1000 : 10); + response.ResponseHeader.RequestHandle = requestId + 100; + if (sentWhileDispatching) + { + _ = ((TcpServerChannel)target).SendResponseAsync(requestId, response).AsTask(); + } + else + { + _ = Task.Run(() => ((TcpServerChannel)target).SendResponseAsync(requestId, response).AsTask()); + } + }); + + for (uint ii = 1; ii <= 3; ii++) + { + transport.Feed(channel.CreateRequestChunkForTest(ii, ii)); + } + channel.StartReceiveLoopForTest(); + + List chunks = await transport.WaitForFinalChunksAsync(3).ConfigureAwait(false); + AssertWireOrder(chunks, [1, 2, 3]); + foreach (WireChunk chunk in chunks) + { + IServiceResponse response = DecodeBody(channel, chunk); + Assert.That(response.ResponseHeader.RequestHandle, Is.EqualTo(chunk.RequestId + 100)); + if (chunk.RequestId == 2) + { + Assert.That(response, Is.InstanceOf()); + // the status of the encoding failure, as the gate path reported it before. + Assert.That(response.ResponseHeader.ServiceResult, Is.EqualTo(StatusCodes.BadEncodingLimitsExceeded)); + } + else + { + Assert.That(response, Is.InstanceOf()); + } + } + + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + + /// + /// A message whose security fails after its chunks took sequence numbers leaves a + /// gap the peer would reject: the writer numbers nothing more for the transport and + /// reports a transport error, which faults the channel. + /// + [Test] + public async Task ASealFailureAfterNumberingStopsTheChannelAsync() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var pool = new CountingPool(); + using SealProbe writer = SealProbe.Create( + SecurityPolicies.Basic256Sha256, MessageSecurityMode.SignAndEncrypt, telemetry, pool); + using SealProbe reader = SealProbe.Create( + SecurityPolicies.Basic256Sha256, MessageSecurityMode.SignAndEncrypt, telemetry, new CountingPool()); + var transport = new RecordingTransport(asyncSendEvery: 2); + writer.SetTransport(transport); + var readerTokens = new Dictionary(); + try + { + SealProbe.Renew(writer, reader, 1, readerTokens); + writer.Send(1, CreateResponse(null, 10)); + await transport.WaitForMessageCountAsync(1).ConfigureAwait(false); + + byte[]? key = writer.BreakEncryptionForTest(); + writer.Send(2, CreateResponse(null, 10)); + await WaitUntilAsync(() => writer.TransportErrorCount > 0).ConfigureAwait(false); + + // a producer numbers a message before the channel faults, as an + // OpenSecureChannel response or a fault does: it follows the gap too. + writer.QueueNumberedMessageForTest(4); + + // sealable again, but the numbers after the gap would be rejected. + writer.RestoreEncryptionForTest(key); + writer.Send(3, CreateResponse(null, 10)); + await WaitUntilAsync(() => !writer.HasPendingWritesForTest).ConfigureAwait(false); + List chunks = await transport.WaitForMessageCountAsync(1).ConfigureAwait(false); + Assert.That(chunks, Has.Count.EqualTo(1), "nothing was sent after the failed message."); + Assert.That(writer.TransportErrorCount, Is.EqualTo(1)); + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + finally + { + reader.ReleaseTokens(); + foreach (ChannelToken token in readerTokens.Values) + { + token.Dispose(); + } + } + } + + [Test] + public async Task ResponsesQueuedForAReplacedTransportAreNotNumberedAsync() + { + // A reconnect replaces the transport and answers on the new one with an + // OpenSecureChannel response numbered under the send order lock. Responses + // still queued for the dropped transport must not take numbers after it: + // the client would see a gap on the new transport. + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var dropped = new RecordingTransport(asyncSendEvery: 1); + dropped.HoldSends(); + channel.SetTransport(dropped); + + await channel.SendResponseAsync(1, CreateResponse(null, 0)).ConfigureAwait(false); + await dropped.FirstSendStarted.ConfigureAwait(false); + for (uint requestId = 2; requestId <= 10; requestId++) + { + await channel.SendResponseAsync(requestId, CreateResponse(null, requestId % 3 == 0 ? 20000 : 0)).ConfigureAwait(false); + } + + var replacement = new RecordingTransport(asyncSendEvery: 2); + channel.SetTransport(replacement); + channel.QueueNumberedMessageForTest(1000); + dropped.ReleaseSends(); + + for (uint requestId = 11; requestId <= 20; requestId++) + { + await channel.SendResponseAsync(requestId, CreateResponse(null, requestId % 3 == 0 ? 20000 : 0)).ConfigureAwait(false); + } + + List chunks = await replacement.WaitForFinalChunksAsync(11).ConfigureAwait(false); + AssertWireOrder(chunks, Enumerable.Range(11, 10).Select(id => (uint)id).Prepend(1000u)); + Assert.That(chunks[0].RequestId, Is.EqualTo(1000u)); + await AssertBuffersReturnedAsync(pool).ConfigureAwait(false); + } + + [Test] + public async Task ReceiveBackpressureEngagesAndReleasesAsync() + { + // OPC UA servers stop reading while more than 100 responses wait to be + // written, and read again once half of them are gone. + const int requests = 180; + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 1); + transport.HoldSends(); + channel.SetTransport(transport); + channel.SetRequestReceivedCallback((target, requestId, request) => + _ = ((TcpServerChannel)target).SendResponseAsync(requestId, CreateResponse(null, 0)).AsTask()); + + for (uint ii = 1; ii <= requests; ii++) + { + transport.Feed(channel.CreateRequestChunkForTest(ii, ii)); + } + channel.StartReceiveLoopForTest(); + + await WaitUntilAsync(() => channel.ChannelFull).ConfigureAwait(false); + + // the loop stops reading: the reads stay where they are. + int readsWhenFull = transport.ReceiveCalls; + await Task.Delay(200).ConfigureAwait(false); + Assert.That(transport.ReceiveCalls, Is.EqualTo(readsWhenFull)); + Assert.That(readsWhenFull, Is.LessThan(requests)); + + transport.ReleaseSends(); + + List chunks = await transport.WaitForFinalChunksAsync(requests).ConfigureAwait(false); + AssertWireOrder(chunks, Enumerable.Range(1, requests).Select(id => (uint)id)); + Assert.That(channel.ChannelFull, Is.False); + } + + [Test] + [Repeat(3)] + public async Task RandomReceiveSendAndRenewalInterleavingKeepsTheWireOrderAsync() + { + // A fuzz-style mix on one channel: requests arrive on the receive loop, each + // is answered from the pool after a random delay, with a random size, while + // tokens are activated and numbered messages are queued. + const int requests = 400; + int seed = Environment.TickCount; + TestContext.Out.WriteLine($"Seed {seed}"); + var random = new Random(seed); + int[] sizes = [.. Enumerable.Range(0, requests + 1).Select(_ => random.Next(4) == 0 ? random.Next(7000, 26000) : random.Next(0, 200))]; + int[] delays = [.. Enumerable.Range(0, requests + 1).Select(_ => random.Next(3))]; + + var pool = new CountingPool(); + using OrderServerChannel channel = OrderServerChannel.Create(pool); + var transport = new RecordingTransport(asyncSendEvery: 4); + channel.SetTransport(transport); + int answered = 0; + channel.SetRequestReceivedCallback((target, requestId, request) => + _ = Task.Run(async () => + { + if (delays[requestId] > 0) + { + await Task.Delay(delays[requestId] - 1).ConfigureAwait(false); + } + await ((TcpServerChannel)target) + .SendResponseAsync(requestId, CreateResponse(null, sizes[requestId])) + .ConfigureAwait(false); + Interlocked.Increment(ref answered); + })); + + channel.StartReceiveLoopForTest(); + + uint tokenId = 1; + uint fed = 0; + uint numberedId = 500_000; + var numberedIds = new List(); + DateTime deadline = DateTime.UtcNow.AddSeconds(60); + while (Volatile.Read(ref answered) < requests && DateTime.UtcNow < deadline) + { + // a request is secured with the newest token when the loop reads it, so + // a renewed token is activated by the receive loop as a client does. + for (int ii = random.Next(1, 8); ii > 0 && fed < requests; ii--) + { + uint requestId = ++fed; + transport.Feed(() => channel.CreateRequestChunkForTest( + requestId, + requestId, + channel.NewestTokenIdForTest)); + } + + // one renewal at a time: the next once the client used the last one. + if (random.Next(3) == 0 && channel.CurrentTokenIdForTest == tokenId) + { + channel.RenewTokenForTest(++tokenId); + } + else + { + channel.QueueNumberedMessageForTest(numberedId); + numberedIds.Add(numberedId++); + } + await Task.Delay(1).ConfigureAwait(false); + } + + Assert.That(Volatile.Read(ref answered), Is.EqualTo(requests)); + List chunks = await transport + .WaitForFinalChunksAsync(requests + numberedIds.Count) + .ConfigureAwait(false); + AssertWireOrder(chunks, Enumerable.Range(1, requests).Select(id => (uint)id).Concat(numberedIds)); + Assert.That(channel.CurrentTokenIdForTest, Is.GreaterThan(1u), "no renewed token was activated."); + + uint lastTokenId = 0; + foreach (WireChunk chunk in chunks) + { + Assert.That(chunk.TokenId, Is.GreaterThanOrEqualTo(lastTokenId)); + lastTokenId = chunk.TokenId; + } + } + + /// + /// Checks that the chunks carry consecutive sequence numbers, that the chunks + /// of a message are contiguous, and that every expected message is complete + /// exactly once. + /// + private static void AssertWireOrder(List chunks, IEnumerable expectedRequestIds) + { + Assert.That(chunks, Is.Not.Empty); + var completed = new HashSet(); + uint? open = null; + for (int ii = 0; ii < chunks.Count; ii++) + { + WireChunk chunk = chunks[ii]; + if (ii > 0) + { + Assert.That( + chunk.SequenceNumber, + Is.EqualTo(chunks[ii - 1].SequenceNumber + 1), + $"chunk {ii} (request {chunk.RequestId}) is out of sequence."); + } + + if (open != null) + { + Assert.That( + chunk.RequestId, + Is.EqualTo(open.Value), + $"chunk {ii} interleaves request {chunk.RequestId} into request {open}."); + } + + if (chunk.IsFinal || chunk.IsAbort) + { + Assert.That(completed.Add(chunk.RequestId), Is.True, $"request {chunk.RequestId} was sent twice."); + open = null; + } + else + { + open = chunk.RequestId; + } + } + + Assert.That(open, Is.Null, "the last message is incomplete."); + Assert.That(completed, Is.EquivalentTo(expectedRequestIds)); + } + + private static async Task AssertBuffersReturnedAsync(CountingPool pool) + { + await WaitUntilAsync(() => pool.OutstandingCount == 0).ConfigureAwait(false); + Assert.That(pool.OutstandingCount, Is.Zero); + Assert.That(pool.DuplicateReturnCount, Is.Zero); + } + + private static async Task WaitUntilAsync(Func condition) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(30); + while (!condition() && DateTime.UtcNow < deadline) + { + await Task.Delay(5).ConfigureAwait(false); + } + Assert.That(condition(), Is.True, "the condition was not reached in time."); + } + + private static async Task RunProducersAsync(int count, Func producer) + { + using var start = new ManualResetEventSlim(); + Task[] tasks = [.. Enumerable.Range(0, count).Select(index => Task.Run(async () => + { + start.Wait(); + await producer(index, new Random(index + 1)).ConfigureAwait(false); + }))]; + start.Set(); + await Task.WhenAll(tasks).ConfigureAwait(false); + } + + /// + /// Decodes the body of a single-chunk message sent on a None channel. + /// + private static IServiceResponse DecodeBody(OrderServerChannel channel, WireChunk chunk) + { + const int headerSize = TcpMessageLimits.SymmetricHeaderSize + TcpMessageLimits.SequenceHeaderSize; + Assert.That(chunk.IsFinal, Is.True); + using var stream = new System.IO.MemoryStream(chunk.Bytes, headerSize, chunk.Bytes.Length - headerSize); + return BinaryDecoder.DecodeMessage(stream, channel.MessageContextForTest); + } + + private static ReadResponse CreateResponse(Random? random, int? payloadSize = null) + { + int size = payloadSize ?? + (random!.Next(4) == 0 ? random.Next(7000, 26000) : random.Next(0, 200)); + return new ReadResponse + { + ResponseHeader = new ResponseHeader + { + ServiceResult = StatusCodes.Good + }, + Results = + [ + new DataValue(new Variant(new ByteString(new byte[size])), StatusCodes.Good) + ] + }; + } + + /// + /// The header fields of one chunk as it was sent. + /// + private sealed record WireChunk( + uint MessageType, + uint TokenId, + uint SequenceNumber, + uint RequestId, + byte[] Bytes) + { + public bool IsFinal => TcpMessageType.IsFinal(MessageType); + + public bool IsAbort => TcpMessageType.IsAbort(MessageType); + } + + /// + /// A server channel opened with a None token, as the buffer tests use. + /// + private sealed class OrderServerChannel : TcpServerChannel + { + private OrderServerChannel( + ITcpChannelListener listener, + BufferManager bufferManager, + ChannelQuotas quotas, + ITelemetryContext telemetry) + : base( + nameof(UaSCBinaryChannelSendOrderTests), + listener, + bufferManager, + quotas, + null!, + [], + telemetry, + new FakeTimeProvider(DateTimeOffset.UtcNow)) + { + } + + public static OrderServerChannel Create(CountingPool pool, int maxByteStringLength = 0) + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + var context = ServiceMessageContext.Create(telemetry); + if (maxByteStringLength > 0) + { + context.MaxByteStringLength = maxByteStringLength; + } + var quotas = new ChannelQuotas(context) + { + MaxBufferSize = 8192, + MaxMessageSize = 4 * 1024 * 1024 + }; + var manager = new BufferManager( + nameof(UaSCBinaryChannelSendOrderTests), + quotas.MaxBufferSize, + telemetry, + pool); + var listener = new Mock(); + listener.SetupGet(value => value.EndpointUrl) + .Returns(new Uri("opc.tcp://localhost:4840")); + var channel = new OrderServerChannel(listener.Object, manager, quotas, telemetry); + channel.ChannelId = 1; + channel.State = TcpChannelState.Open; + channel.LoadTokenForTest(1); + return channel; + } + + public bool HasQueuedWritesForTest => HasPendingWrites; + + public IServiceMessageContext MessageContextForTest => Quotas.MessageContext; + + public void SetTransport(IUaSCByteTransport transport) + { + Transport = transport; + } + + public void StartReceiveLoopForTest() + { + StartReceiveLoop(); + } + + /// + /// Replaces the tokens without the activation path, or removes them. + /// + public void LoadTokenForTest(uint? tokenId) + { + ((IDiagnosticsChannelMutation)this).LoadTokensForOfflineDecode( + tokenId == null ? null : NewToken(tokenId.Value), + previous: null); + } + + public uint CurrentTokenIdForTest => CurrentToken!.TokenId; + + public uint NewestTokenIdForTest => (RenewedToken ?? CurrentToken)!.TokenId; + + /// + /// Issues a renewed token as answering a renewal request does; the receive + /// loop activates it when a request secured with it arrives. + /// + public void RenewTokenForTest(uint tokenId) + { + using (Gate.Enter()) + { + SetRenewedToken(NewToken(tokenId)); + } + } + + /// + /// Queues a message numbered by its producer under the send order lock, + /// the way an asymmetric message is. + /// + public void QueueNumberedMessageForTest(uint requestId) + { + const int size = TcpMessageLimits.SymmetricHeaderSize + TcpMessageLimits.SequenceHeaderSize; + using (SendOrder.Enter()) + { + byte[] buffer = BufferManager.TakeBuffer(size, nameof(QueueNumberedMessageForTest)); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(0), TcpMessageType.Message | TcpMessageType.Final); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(4), size); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(8), ChannelId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(12), CurrentToken!.TokenId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(16), GetNewSequenceNumber()); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(20), requestId); + BeginWriteMessage(new ArraySegment(buffer, 0, size), null); + } + } + + /// + /// Encodes a request chunk the None channel reads. + /// + public ArraySegment CreateRequestChunkForTest(uint requestId, uint sequenceNumber, uint? tokenId = null) + { + using var body = new System.IO.MemoryStream(); + BinaryEncoder.EncodeMessage(new ReadRequest(), body, Quotas.MessageContext, true); + int length = (int)body.Length + 24; + byte[] buffer = BufferManager.TakeBuffer(length, nameof(CreateRequestChunkForTest)); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(0), TcpMessageType.Message | TcpMessageType.Final); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(4), (uint)length); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(8), ChannelId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(12), tokenId ?? CurrentToken!.TokenId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(16), sequenceNumber); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(20), requestId); + body.GetBuffer().AsSpan(0, (int)body.Length).CopyTo(buffer.AsSpan(24)); + return new ArraySegment(buffer, 0, length); + } + + private ChannelToken NewToken(uint tokenId) + { + return new ChannelToken + { + ChannelId = ChannelId, + TokenId = tokenId, + SecurityPolicy = SecurityPolicyInfo.None, + CreatedAt = DateTime.UtcNow, + CreatedAtTimestamp = TimeProvider.GetTimestamp(), + Lifetime = 600000 + }; + } + } + + /// + /// A channel with real keys: one instance writes through the writer, a + /// second one reads what it sent, as the peer does. + /// + private sealed class SealProbe : UaSCUaBinaryChannel + { + private SealProbe( + BufferManager buffers, + ChannelQuotas quotas, + string policyUri, + MessageSecurityMode mode, + ITelemetryContext telemetry) + : base( + "send-order", + buffers, + quotas, + (Certificate?)null, + null, + mode, + policyUri, + telemetry) + { + Buffers = buffers; + ChannelId = 7; + } + + public BufferManager Buffers { get; } + + public static SealProbe Create( + string policyUri, + MessageSecurityMode mode, + ITelemetryContext telemetry, + CountingPool pool) + { + var quotas = new ChannelQuotas(ServiceMessageContext.Create(telemetry)) + { + MaxBufferSize = 8192, + MaxMessageSize = 4 * 1024 * 1024 + }; + var buffers = new BufferManager("send-order", quotas.MaxBufferSize, telemetry, pool); + return new SealProbe(buffers, quotas, policyUri, mode, telemetry); + } + + /// + /// Agrees a token on both sides; the writer activates it while it may be + /// sealing, the reader only keeps it. + /// + public static void Renew( + SealProbe writer, + SealProbe reader, + uint tokenId, + Dictionary readerTokens) + { + byte[] readerNonce = reader.CreateNonce(null)!; + byte[] writerNonce = writer.CreateNonce(null)!; + Assert.That(reader.ValidateNonce(null, writerNonce), Is.True); + Assert.That(writer.ValidateNonce(null, readerNonce), Is.True); + + ChannelToken readerToken = reader.NewToken(tokenId, readerNonce, writerNonce, reader.m_lastSecret); + reader.ComputeKeys(readerToken); + reader.m_lastSecret = readerToken.Secret; + readerTokens[tokenId] = readerToken; + + ChannelToken writerToken = writer.NewToken(tokenId, readerNonce, writerNonce, writer.m_lastSecret); + writer.ActivateToken(writerToken); + writer.m_lastSecret = writerToken.Secret; + } + + public void SetTransport(IUaSCByteTransport transport) + { + Transport = transport; + } + + public uint TakeSequenceNumber() + { + return GetNewSequenceNumber(); + } + + public bool AcceptSequenceNumber(uint sequenceNumber) + { + return VerifySequenceNumber(sequenceNumber, "test"); + } + + public void Send(uint requestId, IServiceResponse response) + { + EncodedSymmetricMessage encoded = EncodeSymmetricMessage( + TcpMessageType.Message, + requestId, + response, + false, + out bool limitsExceeded); + Assert.That(limitsExceeded, Is.False); + Assert.That(BeginWriteSymmetricMessage(encoded, null), Is.True); + } + + /// + /// Verifies and decrypts a chunk with the token it names, and checks its + /// sequence number follows the last one read. + /// + public WireChunk Read(WireChunk chunk, ChannelToken token) + { + ((IDiagnosticsChannelMutation)this).LoadTokensForOfflineDecode(token, previous: null); + byte[] copy = Buffers.TakeBuffer(chunk.Bytes.Length, "test"); + try + { + Buffer.BlockCopy(chunk.Bytes, 0, copy, 0, chunk.Bytes.Length); + ReadSymmetricMessage( + new ArraySegment(copy, 0, chunk.Bytes.Length), + false, + out ChannelToken used, + out uint requestId, + out uint sequenceNumber); + Assert.That(used.TokenId, Is.EqualTo(chunk.TokenId)); + Assert.That( + VerifySequenceNumber(sequenceNumber, "test"), + Is.True, + $"sequence number {sequenceNumber} does not follow on."); + return chunk with { SequenceNumber = sequenceNumber, RequestId = requestId }; + } + finally + { + Buffers.ReturnBuffer(copy, "test"); + } + } + + /// + /// Forgets the tokens the reader borrowed, so disposing it does not + /// dispose them. + /// + /// + /// Number of transport errors the channel reported. + /// + public int TransportErrorCount => Volatile.Read(ref m_transportErrors); + + public bool HasPendingWritesForTest => HasPendingWrites; + + /// + /// Replaces the current token's encrypting key with one AES rejects, so the + /// next message fails after its chunks took their sequence numbers. + /// + /// + /// Numbers and queues a message under the send order lock, as the producers + /// of OpenSecureChannel responses and faults do. + /// + public void QueueNumberedMessageForTest(uint requestId) + { + const int size = TcpMessageLimits.SymmetricHeaderSize + TcpMessageLimits.SequenceHeaderSize; + using (SendOrder.Enter()) + { + byte[] buffer = BufferManager.TakeBuffer(size, nameof(QueueNumberedMessageForTest)); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(0), TcpMessageType.Message | TcpMessageType.Final); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(4), size); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(8), ChannelId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(12), CurrentToken!.TokenId); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(16), GetNewSequenceNumber()); + BinaryPrimitives.WriteUInt32LittleEndian(buffer.AsSpan(20), requestId); + BeginWriteMessage(new ArraySegment(buffer, 0, size), null); + } + } + + public byte[]? BreakEncryptionForTest() + { + byte[]? key = CurrentToken!.ServerEncryptingKey; + CurrentToken!.ServerEncryptingKey = new byte[3]; + return key; + } + + public void RestoreEncryptionForTest(byte[]? key) + { + CurrentToken!.ServerEncryptingKey = key; + } + + protected override void HandleSocketError(ServiceResult result) + { + Interlocked.Increment(ref m_transportErrors); + } + + private int m_transportErrors; + + public void ReleaseTokens() + { + ((IDiagnosticsChannelMutation)this).LoadTokensForOfflineDecode(null, previous: null); + } + + private ChannelToken NewToken(uint tokenId, byte[] clientNonce, byte[] serverNonce, byte[]? previousSecret) + { + ChannelToken token = CreateToken(); + token.TokenId = tokenId; + token.ChannelId = ChannelId; + token.ClientNonce = clientNonce; + token.ServerNonce = serverNonce; + token.PreviousSecret = previousSecret; + token.Lifetime = 600000; + return token; + } + + private byte[]? m_lastSecret; + } + + /// + /// Records every send in order, completing some of them asynchronously, and + /// serves request chunks to a receive loop. + /// + private sealed class RecordingTransport : IUaSCByteTransport, IUaSCChunkCoalescing + { + public RecordingTransport(int asyncSendEvery) + { + m_asyncSendEvery = asyncSendEvery; + } + + public EndPoint LocalEndpoint => null!; + + public EndPoint RemoteEndpoint => null!; + + public TransportChannelFeatures Features => default; + + public string Implementation => "UA-RECORDING"; + + public bool CanCoalesceChunks => true; + + public Task FirstSendStarted => m_firstSendStarted.Task; + + public int ReceiveCalls => Volatile.Read(ref m_receiveCalls); + + public void HoldSends() + { + m_hold = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public void ReleaseSends() + { + m_hold?.TrySetResult(true); + } + + public void Feed(ArraySegment chunk) + { + Feed(() => chunk); + } + + /// + /// Queues a chunk that is created when the receive loop reads it. + /// + public void Feed(Func> chunk) + { + lock (m_lock) + { + m_incoming.Enqueue(chunk); + } + } + + public ValueTask ConnectAsync(Uri url, CancellationToken ct) + { + return default; + } + + public ValueTask SendChunkAsync(ReadOnlyMemory chunk, CancellationToken ct) + { + Record(chunk.ToArray()); + return CompleteAsync(); + } + + public ValueTask SendChunkAsync(BufferCollection buffers, CancellationToken ct) + { + byte[] bytes = new byte[buffers.TotalSize]; + int offset = 0; + foreach (ArraySegment buffer in buffers) + { + Buffer.BlockCopy(buffer.Array!, buffer.Offset, bytes, offset, buffer.Count); + offset += buffer.Count; + } + Record(bytes); + return CompleteAsync(); + } + + public async ValueTask> ReceiveChunkAsync(CancellationToken ct) + { + Interlocked.Increment(ref m_receiveCalls); + while (true) + { + Func>? next = null; + lock (m_lock) + { + if (m_incoming.Count > 0) + { + next = m_incoming.Dequeue(); + } + } + if (next != null) + { + return next(); + } + await Task.Delay(5, ct).ConfigureAwait(false); + } + } + + public void Close() + { + ReleaseSends(); + } + + public void Dispose() + { + Close(); + } + + /// + /// Waits until the given number of final chunks was sent, and returns all + /// chunks sent so far. + /// + public async Task> WaitForFinalChunksAsync(int count) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(60); + while (true) + { + List chunks = Parse(clearHeaders: true); + int finals = chunks.Count(chunk => chunk.IsFinal || chunk.IsAbort); + if (finals >= count || DateTime.UtcNow > deadline) + { + Assert.That(finals, Is.EqualTo(count), "an unexpected number of messages was sent."); + return chunks; + } + await Task.Delay(10).ConfigureAwait(false); + } + } + + /// + /// For encrypted chunks, whose sequence header cannot be read here: waits + /// until the given number of final chunks was sent. + /// + public async Task> WaitForMessageCountAsync(int count) + { + DateTime deadline = DateTime.UtcNow.AddSeconds(60); + while (true) + { + List chunks = Parse(clearHeaders: false); + int finals = chunks.Count(chunk => chunk.IsFinal || chunk.IsAbort); + if (finals >= count || DateTime.UtcNow > deadline) + { + Assert.That(finals, Is.EqualTo(count), "an unexpected number of messages was sent."); + return chunks; + } + await Task.Delay(10).ConfigureAwait(false); + } + } + + private List Parse(bool clearHeaders) + { + byte[] stream; + lock (m_lock) + { + stream = [.. m_sent]; + } + + var chunks = new List(); + int offset = 0; + while (offset + 24 <= stream.Length) + { + uint messageType = BinaryPrimitives.ReadUInt32LittleEndian(stream.AsSpan(offset)); + int size = (int)BinaryPrimitives.ReadUInt32LittleEndian(stream.AsSpan(offset + 4)); + uint tokenId = BinaryPrimitives.ReadUInt32LittleEndian(stream.AsSpan(offset + 12)); + uint sequenceNumber = clearHeaders + ? BinaryPrimitives.ReadUInt32LittleEndian(stream.AsSpan(offset + 16)) + : 0; + uint requestId = clearHeaders + ? BinaryPrimitives.ReadUInt32LittleEndian(stream.AsSpan(offset + 20)) + : 0; + chunks.Add(new WireChunk( + messageType, + tokenId, + sequenceNumber, + requestId, + stream.AsSpan(offset, size).ToArray())); + offset += size; + } + return chunks; + } + + private void Record(byte[] bytes) + { + lock (m_lock) + { + m_sent.AddRange(bytes); + } + m_firstSendStarted.TrySetResult(true); + } + + private ValueTask CompleteAsync() + { + TaskCompletionSource? hold = m_hold; + if (hold != null && !hold.Task.IsCompleted) + { + return new ValueTask(hold.Task); + } + + if (Interlocked.Increment(ref m_sends) % m_asyncSendEvery == 0) + { + return YieldAsync(); + } + return default; + + static async ValueTask YieldAsync() + { + await Task.Yield(); + } + } + + private readonly int m_asyncSendEvery; + private readonly Lock m_lock = new(); + private readonly List m_sent = []; + private readonly Queue>> m_incoming = new(); + private readonly TaskCompletionSource m_firstSendStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private TaskCompletionSource? m_hold; + private int m_sends; + private int m_receiveCalls; + } + + /// + /// Counts outstanding rentals and returns of arrays that were not rented. + /// + private sealed class CountingPool : ArrayPool + { + public override byte[] Rent(int minimumLength) + { + byte[] buffer = new byte[minimumLength]; + lock (m_lock) + { + m_outstanding.Add(buffer); + } + return buffer; + } + + public override void Return(byte[] array, bool clearArray = false) + { + lock (m_lock) + { + if (!m_outstanding.Remove(array)) + { + DuplicateReturnCount++; + } + } + } + + public int OutstandingCount + { + get + { + lock (m_lock) + { + return m_outstanding.Count; + } + } + } + + public int DuplicateReturnCount { get; private set; } + + private readonly Lock m_lock = new(); + private readonly HashSet m_outstanding = []; + } + } +} diff --git a/tests/Opc.Ua.Types.Tests/Utils/ServiceMessageContextMappingTests.cs b/tests/Opc.Ua.Types.Tests/Utils/ServiceMessageContextMappingTests.cs new file mode 100644 index 0000000000..9062c1bee1 --- /dev/null +++ b/tests/Opc.Ua.Types.Tests/Utils/ServiceMessageContextMappingTests.cs @@ -0,0 +1,122 @@ +/* ======================================================================== + * Copyright (c) 2005-2026 The OPC Foundation, Inc. All rights reserved. + * + * OPC Foundation MIT License 1.00 + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the + * Software is furnished to do so, subject to the following + * conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + * OTHER DEALINGS IN THE SOFTWARE. + * + * The complete license agreement can be found here: + * http://opcfoundation.org/License/MIT/1.00/ + * ======================================================================*/ + +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using Opc.Ua.Tests; + +namespace Opc.Ua.Types.Tests.Utils +{ + /// + /// The namespace and server tables of a message context are read without a + /// lock; a reader that compares the table and the mapping version it read + /// earlier with the current ones must still detect every replacement. + /// + [TestFixture] + [Category("Utils")] + [Parallelizable] + public sealed class ServiceMessageContextMappingTests + { + [Test] + public void ReplacingATableAdvancesTheMappingVersion() + { + var context = ServiceMessageContext.CreateEmpty(NUnitTelemetryContext.Create()); + long version = context.MappingVersion; + NamespaceTable first = context.NamespaceUris; + + context.NamespaceUris = first; + Assert.That(context.MappingVersion, Is.EqualTo(version), "setting the same table is no change."); + + var second = new NamespaceTable(); + context.NamespaceUris = second; + Assert.That(context.NamespaceUris, Is.SameAs(second)); + Assert.That(context.MappingVersion, Is.EqualTo(version + 1)); + + var servers = new StringTable(); + context.ServerUris = servers; + Assert.That(context.ServerUris, Is.SameAs(servers)); + Assert.That(context.MappingVersion, Is.EqualTo(version + 2)); + + // back to the first table: the reference matches again, the version does not. + context.NamespaceUris = first; + Assert.That(context.NamespaceUris, Is.SameAs(first)); + Assert.That(context.MappingVersion, Is.EqualTo(version + 3)); + } + + [Test] + public async Task ReadersDetectEveryReplacementWhileTablesAreSwappedAsync() + { + const int swaps = 20000; + var context = ServiceMessageContext.CreateEmpty(NUnitTelemetryContext.Create()); + NamespaceTable a = context.NamespaceUris; + var b = new NamespaceTable(); + long initial = context.MappingVersion; + int done = 0; + int violations = 0; + + Task[] readers = [.. Enumerable.Range(0, 4).Select(_ => Task.Run(() => + { + long lastVersion = initial; + while (Volatile.Read(ref done) == 0) + { + // version, table, version: when the version did not move, the + // table read in between is the one that version belongs to + // (the swaps alternate, so odd versions are b and even ones a). + long before = context.MappingVersion; + NamespaceTable table = context.NamespaceUris; + long after = context.MappingVersion; + if (before < lastVersion || after < before) + { + Interlocked.Increment(ref violations); + } + lastVersion = after; + + if (before == after && + !ReferenceEquals(table, (before - initial) % 2 == 1 ? b : a)) + { + Interlocked.Increment(ref violations); + } + } + }))]; + + for (int ii = 0; ii < swaps; ii++) + { + context.NamespaceUris = ii % 2 == 0 ? b : a; + } + Volatile.Write(ref done, 1); + await Task.WhenAll(readers).ConfigureAwait(false); + + Assert.That(violations, Is.Zero); + Assert.That(context.MappingVersion, Is.EqualTo(initial + swaps)); + Assert.That(context.NamespaceUris, Is.SameAs(a)); + } + } +}