From 5b03e9afa492cdb36049ee52ae844f630f89d8db Mon Sep 17 00:00:00 2001 From: romanett Date: Wed, 14 Oct 2026 10:51:14 +0200 Subject: [PATCH] Channel send path: seal responses in the writer, lock-free gate, fewer locks and allocations per request (#4782) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Carved out of #4771 (roadmap item A1 of the o6 benchmark gap analysis): everything except running requests to completion on the receive loop. #4771 keeps run to completion, stacked on this PR, for a later release. Requests still go through the fair request queue and its workers, as on master. What changes is the channel around them: - the response is encoded without the channel gate; - the channel's single writer numbers, secures and sends; - requests take fewer locks and allocations on the channel. No wire, configuration or public API change. ## Results Paired A/B against master `6cfb8cf19` (with #4767). The builds alternate per round on the same machine, and the table shows medians of 5 rounds with the throughput range of the rounds in brackets. The machine was shared with other work, so the CPU and throughput ranges are wide; allocated bytes per request are deterministic. - **o6read** uses the native open62541 client workers of the o6-automation benchmark, in the o6 2.0.1 configuration: `AsyncCustomNodeManager`, Server GC, queue = clients × outstanding + 3. - **"4 cores"** pins the server to 4 cores. - **The other rows** use the harness's .NET client: 20 sessions unless noted, workstation GC, ReferenceServer. - **Contention** counts runtime lock-contention events in the measured window. | Scenario | req/s | Server CPU µs/req | Alloc B/req | Contention | |---|---|---|---|---| | o6read 10×100, 24 cores | 254,600 → 264,700 (+4 %) [228–292k → 255–270k] | 67.4 → 69.5 | 5,874 → 4,709 (**−20 %**) | 6,766 → 1,811 | | o6read 10×1000, 24 cores | 248,900 → 241,300 (−3 %) [231–258k → 233–260k] | 69.0 → 71.3 | 5,866 → 4,718 (**−20 %**) | 6,080 → 1,456 | | o6read 10×100 encrypted, 24 cores | 203,900 → 232,000 (**+14 %**) [111–226k → 204–264k] | 71.6 → 70.2 | 5,985 → 4,752 (**−21 %**) | 5,166 → 1,288 | | o6read 10×100 `--o6-async 0` (`CustomNodeManager2`), 24 cores | 235,600 → 243,900 (+4 %) | 40.1 → 38.8 | 6,702 → 5,886 (−12 %) | 148k → 124k | | o6read 1 client × 1000 (one channel) | 64,300 → 92,600 (**+44 %**) [52–69k → 76–100k] | 79.8 → 74.5 (−7 %) | 6,003 → 5,063 (−16 %) | 224 → 128 | | o6read 10×100, 4 cores | 156,600 → 140,400 (−10 %, noise: rounds 139–192k → 121–184k) | 25.0 → 28.3 | 5,435 → 4,688 (−14 %) | 97 → 184 | | o6read 10×1000, 4 cores | 107,100 → 125,200 (**+17 %**) [103–116k → 120–136k] | 36.5 → 31.7 (**−13 %**) | 9,966 → 4,767 (**−52 %**) | 312 → 160 | | o6read 10×100 encrypted, 4 cores | 103,300 → 126,300 (**+22 %**) [102–105k → 115–143k] | 37.9 → 31.3 (**−17 %**) | 5,498 → 4,736 (−14 %) | 46 → 29 | | o6read 10×100 `--o6-async 0`, 4 cores | 186,600 → 187,700 (+1 %) | 21.1 → 21.0 | 6,587 → 5,915 (−10 %) | 17.7k → 4.2k | | read --nodes 1, 20 sessions | 208,300 → 205,900 (−1 %) | 49.1 → 48.3 | 4,905 → 4,621 (−6 %) | p50/p99 0.072/1.18 → 0.072/1.26 ms | | read --nodes 1, 200 sessions | 216,400 → 208,600 (−4 %) | 42.0 → 43.2 | 4,790 → 4,518 (−6 %) | p50/p99 0.606/3.56 → 0.615/4.05 ms | | read --nodes 100 | 31,950 → 32,470 (+2 %) | 278 → 253 (−9 %) | 73,702 → 73,436 | p50/p99 0.290/2.51 → 0.283/2.48 ms | | browse | 38,380 → 40,460 (+5 %) | 162 → 161 | 39,305 → 38,981 | p50 0.349 → 0.330 ms | | sub (5 × 100 items, 100 ms) | 88,190 → 89,010 notif/s | 11.1 → 10.7 | 159 → 153 | p50/p99 20.7/64.3 → 20.8/66.0 ms | **Where it helps:** - Allocation per request falls in every scenario, by 20 % for o6read. - Lock contention falls by 70–80 % for o6read. - Throughput rises where the channel gate serialised the work: - one deep pipeline on one channel: +44 %; - encrypted: +14 % on 24 cores, +22 % on 4 cores; - 1000 in flight on 4 cores: +17 %. **Where it does not:** CPU per request at 24 cores is unchanged. The requests still go through the fair queue's workers, and the hand-off to a worker and back is what #4771 removes. The 4-core 10×100 row spread as much within each side as between the sides. ## Changes ### Sending - **Seal in the writer.** The producer encodes the response without the channel gate and queues it unsealed. The channel's single writer takes the queued responses in order under a per-channel send-order lock: it assigns sequence numbers, selects the current token, signs and encrypts, and sends the batch. Messages that number themselves (OpenSecureChannel, asymmetric faults, the client's requests) hold the same lock until they are queued. Token changes that can dispose a token take it as well. - **Reconnect.** The old transport is detached before the new token is activated. Unsealed responses still queued for it are then dropped without taking sequence numbers. The new transport is attached and the OpenSecureChannel response numbered under one hold of the send-order lock, so the response is the first message on the new transport. The channel opens again only after that. - **Failure past the sequence numbers.** A message whose security fails after its chunks took sequence numbers stops all numbering on that transport. The writer drops the messages producers numbered for it after the gap, and reports a transport error once it has left the send-order lock. That faults the channel (server) or reconnects (client). - **Numbered writes without the lock.** The writer takes writes that their producers numbered without the send-order lock, because they are queued in number order. A client channel queues only such writes, so its writer no longer waits for the lock that its request threads take. - **Writer on the pool.** The writer always starts on the thread pool. Running it on the producer's thread sent batches of one message. - **Buffers.** The encoded message is a struct in the write-queue entry. A small single-chunk message moves to a right-sized buffer only when writes are already queued, so deep pipelines no longer exhaust the bounded buffer pool (−52 % allocation at 10×1000 on 4 cores). ### Locks and allocations per request - **`ChannelGate`** takes the gate with one compare-and-exchange when it is free, and queues waiters under a lock only when it is not. `SemaphoreSlim` took its monitor on every wait and release, which the receive loop does for every message. The contract is unchanged: not re-entrant, FIFO handoff, cancellable waits, synchronous completion when free. The transport's send lock is a `ChannelGate` too. - **Namespace and server tables.** Every encoder and decoder reads `ServiceMessageContext.NamespaceUris`, `ServerUris` and `MappingVersion`. They are now read lock-free from one immutable snapshot that setters swap atomically. - **Allocation-free tracking.** Tracking a pending request (`PendingRequestScope`) and handing a decoded request out of the gate allocate nothing. - **Single-chunk decode.** A request in one chunk is decoded from that chunk's buffer. - **Fair-queue entry.** An entry coordinates its cancellation registration with an interlocked state (`RegistrationState`) instead of a lock object. ### Client fix `Session.CloseAsync` disposed the session synchronously, and `Dispose` waits for the async dispose of the session and its subscription engine. Closing 1000 sessions at once blocked one pool thread per session while that dispose needed pool threads, and the close took minutes in `ServerManySessionsLoadTestAsync(1000)`. `CloseAsync` now awaits `DisposeAsync`, and dispose awaits an engine that implements `IAsyncDisposable`, as `DefaultSubscriptionEngine` now does. ## Behaviour notes - **Wire order and tokens.** Sequence numbers stay strictly increasing in wire order, and the chunks of one message stay contiguous. Token renewal applies at a message boundary. After a reconnect, the OpenSecureChannel response is the first message on the new transport. - **Back-pressure** (#4707) is unchanged. The receive loop stops reading above 100 queued writes. - **Client.** `Session.CloseAsync` no longer blocks a thread on the dispose. ## Not in this PR The striped server-stop tokens of #4771 are dropped. Since #4767 a server request no longer registers with the stop token, because the deadline monitor polls it. ## Testing - `UaSCBinaryChannelSendOrderTests` covers: - wire order and random interleaving; - token renewal (decrypt and verify) for five policies; - buffers returned once; - a seal failure after numbering stops the channel, and a message numbered after the gap is not sent; - the replaced transport; - back-pressure; - a response that cannot be encoded is answered with a fault. - `TcpReconnectKeyAgreementTests.AResponseQueuedForTheDroppedTransportIsNotSecuredWithTheNewTokenAsync` (new) holds the old transport's sends during a reconnect. It fails on the previous order. The seal-failure test fails without the gap check. - Also run: `ChannelGateTests`, `TcpByteTransportTests`, `ServiceMessageContextMappingTests`. - Local, net10.0: - Opc.Ua.Core.Tests (Stack.Server, Transport, Tcp, ChannelGate): 2739 passed, 0 failed; - Opc.Ua.Configuration.Tests: 241 passed, 0 failed. net48 builds. `dotnet format whitespace --verify-no-changes` passes on the changed files. - CI: the `CI` workflow on this head is running; the commits were tested together with run to completion in #4771 (its last full CI run passed). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../Tcp/KestrelTcpTransportListener.cs | 2 +- src/Opc.Ua.Client/Session/Session.cs | 15 +- .../Subscription/DefaultSubscriptionEngine.cs | 12 +- .../Stack/Server/FairRequestQueue.cs | 69 +- src/Opc.Ua.Core/Stack/Tcp/ChannelGate.cs | 285 +++- src/Opc.Ua.Core/Stack/Tcp/TcpByteTransport.cs | 25 +- .../Stack/Tcp/TcpListenerChannel.cs | 67 +- src/Opc.Ua.Core/Stack/Tcp/TcpServerChannel.cs | 271 ++-- .../Stack/Tcp/TcpTransportListener.cs | 2 +- .../Stack/Tcp/UaSCBinaryChannel.Symmetric.cs | 305 +++- .../Stack/Tcp/UaSCBinaryChannel.cs | 401 +++++- .../Utils/ServiceMessageContext.cs | 67 +- .../Stack/Transport/TcpByteTransportTests.cs | 12 +- .../TcpReconnectKeyAgreementTests.cs | 93 +- .../Transport/TcpServerChannelBufferTests.cs | 17 +- .../UaSCBinaryChannelSendOrderTests.cs | 1267 +++++++++++++++++ .../ServiceMessageContextMappingTests.cs | 122 ++ 17 files changed, 2732 insertions(+), 300 deletions(-) create mode 100644 tests/Opc.Ua.Core.Tests/Stack/Transport/UaSCBinaryChannelSendOrderTests.cs create mode 100644 tests/Opc.Ua.Types.Tests/Utils/ServiceMessageContextMappingTests.cs 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)); + } + } +}